fix(parquet): three pre-existing defects — unreadable FLBA files, truncated bounds, an ungated annotation

src/EngineeredWood.Parquet/Parquet/Data/ArrowSchemaConverter.cs

@@ -303,7 +303,13 @@ public static IArrowType ToArrowType(ColumnDescriptor column, ParquetReadOptions
                 { BitWidth: 64, IsSigned: false } => UInt64Type.Default,
                 _ => null,
             },
-            LogicalType.TimestampType ts => new TimestampType(
+            // The guard is load-bearing, not defensive. This arm hands back an Arrow TimestampType, and
+            // the read path then reinterprets the column's value buffer as int64 (ArrowArrayBuilder maps
+            // `Int64Type or TimestampType or Time64Type` to a long buffer). A TIMESTAMP annotation on any
+            // other physical width therefore decoded silent garbage instead of failing. Falling through
+            // to the physical type is lossless. MakeDecimalType has always switched on the physical type
+            // for the same reason; this arm had not.
+            LogicalType.TimestampType ts when column.PhysicalType == PhysicalType.Int64 => new TimestampType(
                 ts.Unit switch
                 {
                     Metadata.TimeUnit.Millis => Apache.Arrow.Types.TimeUnit.Millisecond,
@@ -344,8 +350,14 @@ public static IArrowType ToArrowType(ColumnDescriptor column, ParquetReadOptions
             // TimeZoneInfo overload renders the "+00:00" offset. A file old enough to carry only a
             // converted type is exactly the kind this matters for, since it was written by a tool
             // whose output is still being read by everything else.
-            ConvertedType.TimestampMillis => new TimestampType(Apache.Arrow.Types.TimeUnit.Millisecond, "UTC"),
-            ConvertedType.TimestampMicros => new TimestampType(Apache.Arrow.Types.TimeUnit.Microsecond, "UTC"),
+            // Guarded on INT64 for the same reason as the logical-type arm: these hand back an Arrow
+            // TimestampType, whose buffer is read as int64. TIMESTAMP_MILLIS / TIMESTAMP_MICROS are
+            // INT64-only converted types, so a file carrying one on another width is malformed — and
+            // reading it as raw bytes beats decoding a plausible-looking wrong date.
+            ConvertedType.TimestampMillis when column.PhysicalType == PhysicalType.Int64
+                => new TimestampType(Apache.Arrow.Types.TimeUnit.Millisecond, "UTC"),
+            ConvertedType.TimestampMicros when column.PhysicalType == PhysicalType.Int64
+                => new TimestampType(Apache.Arrow.Types.TimeUnit.Microsecond, "UTC"),
             ConvertedType.TimeMillis => new Time32Type(Apache.Arrow.Types.TimeUnit.Millisecond),
             ConvertedType.TimeMicros => new Time64Type(Apache.Arrow.Types.TimeUnit.Microsecond),
             ConvertedType.Int8 => Int8Type.Default,

src/EngineeredWood.Parquet/Parquet/Data/ColumnChunkReader.cs

@@ -1183,7 +1183,7 @@ private static void DecodeDeltaByteArrayValues(
             throw new NotSupportedException(
                 $"Physical type '{column.PhysicalType}' is not supported for DELTA_BYTE_ARRAY decoding.");
 
-        DeltaByteArrayDecoder.Decode(data, count, state);
+        DeltaByteArrayDecoder.Decode(data, count, state, column.TypeLength ?? 0);
     }
 
     private static void DecodeByteStreamSplitValues(

src/EngineeredWood.Parquet/Parquet/Data/ColumnChunkWriter.cs

@@ -233,7 +233,7 @@ private static ColumnChunkResult WriteColumnCore(
                 : StatisticsCollector.Compute(
                     array, physicalType, typeLength, valueDefLevels, nonNullCount, rowCount,
                     floatingPointTotalOrder);
-            result.MetaData.Statistics = DropDeprecatedMinMaxIfMisordered(stats, ValueType(array));
+            result.MetaData.Statistics = DropDeprecatedMinMaxIfMisordered(stats, ValueType(array), physicalType);
         }
 
         // Build Bloom filter if enabled for this column.
@@ -502,7 +502,7 @@ internal static ColumnChunkResult WriteDictionaryColumnFromResult(
                 dictResult.DictionaryPageData, dictResult.DictionaryCount,
                 physicalType, typeLength, rowCount - nonNullCount);
 
-        result.MetaData.Statistics = DropDeprecatedMinMaxIfMisordered(stats, arrowType);
+        result.MetaData.Statistics = DropDeprecatedMinMaxIfMisordered(stats, arrowType, physicalType);
         return result;
     }
 
@@ -1667,9 +1667,9 @@ private static int EstimateColumnSize(int rowCount, PhysicalType physicalType, i
     /// signed ints incl. date/time/timestamp, floats); drop them elsewhere.
     /// </remarks>
     private static Statistics DropDeprecatedMinMaxIfMisordered(
-        Statistics stats, Apache.Arrow.Types.IArrowType type)
+        Statistics stats, Apache.Arrow.Types.IArrowType type, PhysicalType physicalType)
     {
-        if (SignedOrderMatchesLogical(type))
+        if (SignedOrderMatchesLogical(type, physicalType))
             return stats;
 
         return new Statistics
@@ -1686,12 +1686,31 @@ private static Statistics DropDeprecatedMinMaxIfMisordered(
 
     // True when the type's SIGNED byte ordering equals its logical ordering — the precondition for emitting
     // the deprecated Statistics.min/max fields (see DropDeprecatedMinMaxIfMisordered).
-    private static bool SignedOrderMatchesLogical(Apache.Arrow.Types.IArrowType type) => type switch
+    //
+    // The real precondition is narrower than "this Arrow type is signed": it is that StatisticsCollector
+    // compared these values with a TYPED comparator. It does that only for BOOLEAN/INT32/INT64/FLOAT/DOUBLE;
+    // every FIXED_LEN_BYTE_ARRAY and BYTE_ARRAY column goes through SequenceCompareTo, i.e. unsigned
+    // lexicographic. So the physical type has to be part of the answer wherever an Arrow type can arrive on
+    // more than one physical width.
+    // Internal rather than private so the gate can be pinned directly. The FIXED_LEN_BYTE_ARRAY answer is
+    // latent until an Arrow TimestampType can map to that physical type, so there is no end-to-end write
+    // that reaches it yet — a unit test is the only thing that keeps the fix from silently regressing.
+    internal static bool SignedOrderMatchesLogical(
+        Apache.Arrow.Types.IArrowType type, PhysicalType physicalType) => type switch
     {
         BooleanType => true,
         Int8Type or Int16Type or Int32Type or Int64Type => true,
-        FloatType or DoubleType or HalfFloatType => true,
-        Date32Type or Date64Type or Time32Type or Time64Type or TimestampType or DurationType => true,
+        FloatType or DoubleType => true,
+        // FLOAT16 is FLBA(2), so this claim is already wrong for it — its bounds come from a lexicographic
+        // byte comparison over IEEE-754 halves. That is a separately tracked gap (doc/known-issues.md), and
+        // is left exactly as it was rather than quietly changed under cover of this fix.
+        HalfFloatType => true,
+        // Date/Time/Duration only ever arrive on INT32/INT64, so the collector used a typed comparator.
+        Date32Type or Date64Type or Time32Type or Time64Type or DurationType => true,
+        // TIMESTAMP is the one that can also arrive on FIXED_LEN_BYTE_ARRAY. There the collector compares
+        // bytes unsigned-lexicographically, which is not the signed order these deprecated fields promise —
+        // and a wrong bound in the footer is a wrong prune, not a cosmetic defect.
+        TimestampType => physicalType == PhysicalType.Int64,
         Decimal32Type or Decimal64Type => true, // INT32/INT64 physical — signed numeric ordering
         _ => false, // UTF-8 strings/binary (unsigned lexical), unsigned ints, decimal FLBA, nested, ...
     };

src/EngineeredWood.Parquet/Parquet/Data/DeltaByteArrayDecoder.cs

@@ -19,8 +19,18 @@ internal static class DeltaByteArrayDecoder
     /// <summary>
     /// Decodes <paramref name="count"/> byte array values and appends them to <paramref name="state"/>.
     /// </summary>
-    public static void Decode(ReadOnlySpan<byte> data, int count, ColumnBuildState state)
+    /// <param name="typeLength">
+    /// Fixed value width for a FIXED_LEN_BYTE_ARRAY column, or 0 for BYTE_ARRAY. The two physical types
+    /// share this encoding but not their destination buffers, and the state only allocates the pair the
+    /// column's physical type calls for -- so the width has to reach here rather than be inferred.
+    /// </param>
+    public static void Decode(ReadOnlySpan<byte> data, int count, ColumnBuildState state, int typeLength = 0)
     {
+        bool fixedWidth = state.PhysicalType == PhysicalType.FixedLenByteArray;
+        if (fixedWidth && typeLength <= 0)
+            throw new ParquetFormatException(
+                "A FIXED_LEN_BYTE_ARRAY column decoded as DELTA_BYTE_ARRAY has no type_length.");
+
         // Step 1: Decode prefix lengths
         var prefixDecoder = new DeltaBinaryPackedDecoder(data);
         var prefixLengths = new int[count];
@@ -38,14 +48,51 @@ public static void Decode(ReadOnlySpan<byte> data, int count, ColumnBuildState s
         // Step 4: Reconstruct values by combining prefix from previous value + suffix
         // Compute total output size
         var valueLengths = new int[count];
-        int totalBytes = 0;
+        long totalBytes = 0;
+        long totalSuffixBytes = 0;
         for (int i = 0; i < count; i++)
         {
-            valueLengths[i] = prefixLengths[i] + suffixLengths[i];
+            int prefixLen = prefixLengths[i];
+            int suffixLen = suffixLengths[i];
+
+            if (prefixLen < 0 || suffixLen < 0)
+                throw new ParquetFormatException(
+                    $"DELTA_BYTE_ARRAY value at index {i} has a negative prefix ({prefixLen}) or suffix " +
+                    $"({suffixLen}) length.");
+
+            // A value is the first prefixLen bytes of the PREVIOUS value plus a suffix. A prefix that does
+            // not fit inside the previous value -- including ANY prefix on the first value, which has no
+            // predecessor -- would be reconstructed from the zero-filled bytes reserved for this value:
+            // neither what was encoded nor an error. The output buffer is sized from these same lengths,
+            // so nothing reads out of bounds and nothing throws; it just comes out wrong.
+            int previousLength = i == 0 ? 0 : valueLengths[i - 1];
+            if (prefixLen > previousLength)
+                throw new ParquetFormatException(
+                    $"DELTA_BYTE_ARRAY value at index {i} claims a {prefixLen}-byte prefix of a value that " +
+                    $"is {previousLength} bytes long.");
+
+            valueLengths[i] = prefixLen + suffixLen;
+            if (fixedWidth && valueLengths[i] != typeLength)
+                throw new ParquetFormatException(
+                    $"DELTA_BYTE_ARRAY value at index {i} is {valueLengths[i]} bytes, but the column is " +
+                    $"FIXED_LEN_BYTE_ARRAY({typeLength}). Every value in such a column is exactly that wide.");
             totalBytes += valueLengths[i];
+            totalSuffixBytes += suffixLen;
         }
 
-        var outputData = new byte[totalBytes];
+        // Accumulated as long: prefixes let the total grow faster than the page does, so a malformed
+        // page can describe more output than an int can hold.
+        if (totalBytes > int.MaxValue)
+            throw new ParquetFormatException(
+                $"DELTA_BYTE_ARRAY page describes {totalBytes} bytes of values, which exceeds the maximum " +
+                "addressable buffer.");
+
+        if (totalSuffixBytes > rawSuffixes.Length)
+            throw new ParquetFormatException(
+                $"DELTA_BYTE_ARRAY page declares {totalSuffixBytes} suffix bytes but carries " +
+                $"{rawSuffixes.Length}.");
+
+        var outputData = new byte[(int)totalBytes];
         var offsets = new int[count + 1];
         int outputPos = 0;
         int suffixPos = 0;
@@ -73,6 +120,17 @@ public static void Decode(ReadOnlySpan<byte> data, int count, ColumnBuildState s
         }
         offsets[count] = outputPos;
 
+        if (fixedWidth)
+        {
+            // Every value is exactly typeLength bytes, so the reconstruction above is already the packed
+            // layout the fixed-width buffer wants and the offsets are redundant. AddByteArrayValues is NOT
+            // an option here: it writes through the data/offsets buffer pair, which ColumnBuildState only
+            // allocates for BYTE_ARRAY columns -- reaching it with a FIXED_LEN_BYTE_ARRAY column threw a
+            // NullReferenceException.
+            outputData.AsSpan(0, count * typeLength).CopyTo(state.ReserveFixedBytes(count, typeLength));
+            return;
+        }
+
         state.AddByteArrayValues(offsets, outputData, count);
     }
 }

src/EngineeredWood.Parquet/Parquet/ParquetStatisticsAccessor.cs

@@ -106,13 +106,13 @@ private static Dictionary<string, int> BuildNameIndex(SchemaDescriptor schema)
     private static LiteralValue? DecodeMin(ColumnDescriptor desc, Statistics stats)
     {
         var bytes = stats.MinValue ?? FallbackBytes(desc, stats.Min);
-        return bytes is null ? null : Decode(desc, bytes);
+        return bytes is null ? null : Decode(desc, bytes, isMax: false);
     }
 
     private static LiteralValue? DecodeMax(ColumnDescriptor desc, Statistics stats)
     {
         var bytes = stats.MaxValue ?? FallbackBytes(desc, stats.Max);
-        return bytes is null ? null : Decode(desc, bytes);
+        return bytes is null ? null : Decode(desc, bytes, isMax: true);
     }
 
     /// <summary>
@@ -134,7 +134,12 @@ PhysicalType.Int32 or PhysicalType.Int64
         };
     }
 
-    private static LiteralValue? Decode(ColumnDescriptor desc, byte[] bytes)
+    /// <param name="isMax">
+    /// Which end of the range this bound is. It matters wherever the decode cannot be exact: a bound
+    /// must only ever move OUTWARD. Rounding a max down, or a min up, narrows the range the file claims
+    /// and lets a row group be pruned that genuinely contains matching rows.
+    /// </param>
+    private static LiteralValue? Decode(ColumnDescriptor desc, byte[] bytes, bool isMax)
     {
         var logical = desc.SchemaElement.LogicalType;
 
@@ -143,11 +148,11 @@ PhysicalType.Int32 or PhysicalType.Int64
             PhysicalType.Boolean => bytes.Length >= 1
                 ? (LiteralValue?)LiteralValue.Of(bytes[0] != 0) : null,
             PhysicalType.Int32 => DecodeInt32(bytes, logical),
-            PhysicalType.Int64 => DecodeInt64(bytes, logical),
+            PhysicalType.Int64 => DecodeInt64(bytes, logical, isMax),
             PhysicalType.Float => DecodeFloat(bytes),
             PhysicalType.Double => DecodeDouble(bytes),
             PhysicalType.ByteArray => DecodeByteArray(bytes, logical),
-            PhysicalType.FixedLenByteArray => DecodeFixedLenByteArray(desc, bytes, logical),
+            PhysicalType.FixedLenByteArray => DecodeFixedLenByteArray(desc, bytes, logical, isMax),
             // INT96 sort order is undefined per the Parquet spec.
             PhysicalType.Int96 => null,
             _ => null,
@@ -203,7 +208,7 @@ PhysicalType.Int32 or PhysicalType.Int64
         }
     }
 
-    private static LiteralValue? DecodeInt64(byte[] bytes, LogicalType? logical)
+    private static LiteralValue? DecodeInt64(byte[] bytes, LogicalType? logical, bool isMax)
     {
         if (bytes.Length < 8) return null;
         long v = BinaryPrimitives.ReadInt64LittleEndian(bytes);
@@ -215,24 +220,18 @@ PhysicalType.Int32 or PhysicalType.Int64
             case LogicalType.DecimalType d:
                 return LiteralValue.HighPrecisionDecimalOf(new BigInteger(v), d.Scale);
             case LogicalType.TimestampType ts:
-                long unixMs = ts.Unit switch
-                {
-                    TimeUnit.Millis => v,
-                    TimeUnit.Micros => v / 1000,
-                    TimeUnit.Nanos => v / 1_000_000,
-                    _ => v,
-                };
-                var dto = ts.IsAdjustedToUtc
-                    ? DateTimeOffset.FromUnixTimeMilliseconds(unixMs)
-                    : new DateTimeOffset(
-                        DateTimeOffset.FromUnixTimeMilliseconds(unixMs).Ticks,
-                        TimeSpan.Zero);
-                return LiteralValue.Of(dto);
+                return TimestampLiteral(new BigInteger(v), ts, isMax);
 #if NET6_0_OR_GREATER
             case LogicalType.TimeType t when t.Unit == TimeUnit.Micros:
-                return LiteralValue.Of(new TimeOnly(v * 10)); // micros → ticks (10 ticks per us)
+                return LiteralValue.Of(new TimeOnly(v * 10)); // micros → ticks (10 ticks per us), exact
             case LogicalType.TimeType t when t.Unit == TimeUnit.Nanos:
-                return LiteralValue.Of(new TimeOnly(v / 100)); // 100 ns per tick
+                // 100 ns per tick, so this is the one TIME unit that cannot be exact. Round outward, then
+                // clamp: rounding the last nanoseconds of a day up would leave TimeOnly's range, and its
+                // maximum is still a sound outward bound.
+                long timeTicks = (long)DivideOutward(v, 100, isMax);
+                if (timeTicks < 0 || timeTicks > TimeOnly.MaxValue.Ticks)
+                    timeTicks = Math.Clamp(timeTicks, 0, TimeOnly.MaxValue.Ticks);
+                return LiteralValue.Of(new TimeOnly(timeTicks));
 #endif
             default:
                 return LiteralValue.Of(v);
@@ -254,7 +253,7 @@ or LogicalType.JsonType
     }
 
     private static LiteralValue? DecodeFixedLenByteArray(
-        ColumnDescriptor desc, byte[] bytes, LogicalType? logical)
+        ColumnDescriptor desc, byte[] bytes, LogicalType? logical, bool isMax)
     {
         switch (logical)
         {
@@ -274,6 +273,58 @@ or LogicalType.JsonType
         }
     }
 
+    /// <summary>Ticks (100 ns) from .NET's epoch (0001-01-01) to the Unix epoch.</summary>
+    private const long UnixEpochTicks = 621_355_968_000_000_000L;
+
+    /// <summary>
+    /// Builds a timestamp bound from a count of <paramref name="ts"/>'s unit since the Unix epoch.
+    /// </summary>
+    /// <remarks>
+    /// <para>Goes through TICKS rather than milliseconds. A <see cref="DateTimeOffset"/> holds 100 ns,
+    /// so MILLIS and MICROS convert exactly and only NANOS has to round -- where the previous
+    /// millisecond conversion threw away everything below a millisecond for all three. That was not
+    /// merely imprecise: it truncated toward zero, so a max bound of 1500 us came back as 0 ms, and a
+    /// row group whose rows genuinely matched `t > 0.5ms` could be pruned on the strength of it.</para>
+    ///
+    /// <para>Returns <see langword="null"/> outside <see cref="DateTimeOffset"/>'s range. A bound that
+    /// cannot be represented is not a bound; clamping one would be indistinguishable from a real
+    /// endpoint and would prune on a value the file never contained.</para>
+    /// </remarks>
+    private static LiteralValue? TimestampLiteral(BigInteger value, LogicalType.TimestampType ts, bool isMax)
+    {
+        BigInteger unixTicks = ts.Unit switch
+        {
+            TimeUnit.Millis => value * 10_000,
+            TimeUnit.Micros => value * 10,
+            TimeUnit.Nanos => DivideOutward(value, 100, isMax),
+            _ => value * 10,
+        };
+
+        BigInteger ticks = unixTicks + UnixEpochTicks;
+        if (ticks < BigInteger.Zero || ticks > DateTime.MaxValue.Ticks)
+            return null;
+
+        return LiteralValue.Of(new DateTimeOffset((long)ticks, TimeSpan.Zero));
+    }
+
+    /// <summary>
+    /// Divides so the result moves AWAY from zero-error: a max bound rounds up, a min bound rounds down.
+    /// Both widen the range the bound describes, which is the only safe direction for pruning.
+    /// </summary>
+    private static BigInteger DivideOutward(BigInteger value, int divisor, bool isMax)
+    {
+        // DivRem truncates toward zero, so which adjustment is needed depends on the sign: for a
+        // positive value the quotient is already the floor, for a negative one it is already the ceiling.
+        BigInteger quotient = BigInteger.DivRem(value, divisor, out BigInteger remainder);
+        if (remainder.IsZero)
+            return quotient;
+
+        if (isMax)
+            return value.Sign > 0 ? quotient + BigInteger.One : quotient;
+
+        return value.Sign < 0 ? quotient - BigInteger.One : quotient;
+    }
+
     /// <summary>Days from .NET epoch (0001-01-01) to Unix epoch (1970-01-01).</summary>
     private const int EpochDays = 719_162;