Raise exception when reading non-UTC adjusted timestamps in parquet (#…

…4421)

Also moved all parquet files in core repo to Git LFS

.gitattributes

@@ -0,0 +1 @@
+*.parquet filter=lfs diff=lfs merge=lfs -text

.github/workflows/check-ci.yml

@@ -16,6 +16,8 @@ jobs:
  steps:
  - name: Checkout
  uses: actions/checkout@v4
+ with:
+ lfs: true
 
  - name: Setup JDK 11
  id: setup-java-11

.github/workflows/nightly-check-ci.yml

@@ -24,6 +24,8 @@ jobs:
  steps:
  - name: Checkout
  uses: actions/checkout@v4
+ with:
+ lfs: true
 
  - name: Setup JDK 11
  id: setup-java-11

extensions/parquet/base/src/main/java/io/deephaven/parquet/base/ParquetFileReader.java

@@ -236,8 +236,17 @@ private static void buildChildren(Types.GroupBuilder builder, Iterator<SchemaEle
  }
 
  if (schemaElement.isSetLogicalType()) {
- ((Types.Builder) childBuilder)
- .as(getLogicalTypeAnnotation(schemaElement.logicalType));
+ LogicalType logicalType = schemaElement.logicalType;
+ if (logicalType.isSetTIMESTAMP()) {
+ TimestampType timestamp = logicalType.getTIMESTAMP();
+ if (!timestamp.isAdjustedToUTC) {
+ // TODO(deephaven-core#976): Unable to read non UTC adjusted timestamps
+ throw new ParquetFileReaderException(String.format(
+ "Only UTC timestamp is supported, found time column `%s` with isAdjustedToUTC=false",
+ schemaElement.getName()));
+ }
+ }
+ ((Types.Builder) childBuilder).as(getLogicalTypeAnnotation(logicalType));
  }
 
  if (schemaElement.isSetConverted_type()) {
@@ -365,12 +374,16 @@ private static LogicalTypeAnnotation getLogicalTypeAnnotation(ConvertedType type
  case DATE:
  return LogicalTypeAnnotation.dateType();
  case TIME_MILLIS:
+ // TODO(deephaven-core#976) Assuming that time is adjusted to UTC
  return LogicalTypeAnnotation.timeType(true, LogicalTypeAnnotation.TimeUnit.MILLIS);
  case TIME_MICROS:
+ // TODO(deephaven-core#976) Assuming that time is adjusted to UTC
  return LogicalTypeAnnotation.timeType(true, LogicalTypeAnnotation.TimeUnit.MICROS);
  case TIMESTAMP_MILLIS:
+ // TODO(deephaven-core#976) Assuming that time is adjusted to UTC
  return LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS);
  case TIMESTAMP_MICROS:
+ // TODO(deephaven-core#976) Assuming that time is adjusted to UTC
  return LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MICROS);
  case INTERVAL:
  return LogicalTypeAnnotation.IntervalLogicalTypeAnnotation.getInstance();

extensions/parquet/table/src/main/java/io/deephaven/parquet/table/ParquetSchemaReader.java

@@ -351,7 +351,7 @@ public Optional<Class<?>> visit(final LogicalTypeAnnotation.TimeLogicalTypeAnnot
  @Override
  public Optional<Class<?>> visit(
  final LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestampLogicalType) {
- // TODO(deephaven-core#3588): Unable to read parquet TimestampLogicalTypeAnnotation that is not adjusted
+ // TODO(deephaven-core#976): Unable to read parquet TimestampLogicalTypeAnnotation that is not adjusted
  // to UTC
  if (timestampLogicalType.isAdjustedToUTC()) {
  switch (timestampLogicalType.getUnit()) {

extensions/parquet/table/src/main/java/io/deephaven/parquet/table/location/ParquetColumnLocation.java

@@ -700,6 +700,8 @@ private static class LogicalTypeVisitor<ATTR extends Any>
  @Override
  public Optional<ToPage<ATTR, ?>> visit(
  LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestampLogicalType) {
+ // TODO(deephaven-core#976): Unable to read parquet TimestampLogicalTypeAnnotation that is not adjusted
+ // to UTC
  if (timestampLogicalType.isAdjustedToUTC()) {
  return Optional
  .of(ToInstantPage.create(componentType, timestampLogicalType.getUnit()));

extensions/parquet/table/src/test/e0.py

@@ -8,7 +8,8 @@
  "c": np.arange(3, 6).astype("u1"),
  "d": np.arange(4.0, 7.0, dtype="float64"),
  "e": [True, False, True],
- "f": pd.date_range("20130101", periods=3),
+ # TODO(deephaven-core#976): Unable to read parquet TimestampLogicalTypeAnnotation that is not adjusted to UTC
+ # "f": pd.date_range("20130101", periods=3),
  "g": pd.date_range("20130101", periods=3, tz="US/Eastern"),
  "h": pd.Categorical(list("abc")),
  "i": pd.Categorical(list("abc"), ordered=True),

extensions/parquet/table/src/test/e0.requirements.txt

@@ -1,6 +1,6 @@
 numpy==1.24.2
 pandas==1.5.3
-pyarrow==4.0.1
+pyarrow==5.0.0
 python-dateutil==2.8.2
 pytz==2022.7.1
 six==1.16.0

extensions/parquet/table/src/test/e1.py

@@ -8,7 +8,8 @@
  "c": np.arange(3, 6).astype("u1"),
  "d": np.arange(4.0, 7.0, dtype="float64"),
  "e": [True, False, True],
- "f": pd.date_range("20130101", periods=3),
+ # TODO(deephaven-core#976): Unable to read parquet TimestampLogicalTypeAnnotation that is not adjusted to UTC
+ # "f": pd.date_range("20130101", periods=3),
  "g": pd.date_range("20130101", periods=3, tz="US/Eastern"),
  "h": pd.Categorical(list("abc")),
  "i": pd.Categorical(list("abc"), ordered=True),

extensions/parquet/table/src/test/e2.py

@@ -8,7 +8,8 @@
  "c": np.arange(3, 6).astype("u1"),
  "d": np.arange(4.0, 7.0, dtype="float64"),
  "e": [True, False, True],
- "f": pd.date_range("20130101", periods=3),
+ # TODO(deephaven-core#976): Unable to read parquet TimestampLogicalTypeAnnotation that is not adjusted to UTC
+ # "f": pd.date_range("20130101", periods=3),
  "g": pd.date_range("20130101", periods=3, tz="US/Eastern"),
  "h": pd.Categorical(list("abc")),
  "i": pd.Categorical(list("abc"), ordered=True),

extensions/parquet/table/src/test/java/io/deephaven/parquet/table/TestParquetTools.java

@@ -458,24 +458,23 @@ public void e1() {
  assertTableEquals(uncompressed, zstd);
  }
 
- // TODO(deephaven-core#3588): Unable to read parquet TimestampLogicalTypeAnnotation that is not adjusted to UTC
- // @Test
- // public void e2() {
- // final Table uncompressed =
- // ParquetTools.readTable(TestParquetTools.class.getResource("/e2/uncompressed.parquet").getFile());
- //
- // final Table gzip = ParquetTools.readTable(TestParquetTools.class.getResource("/e2/gzip.parquet").getFile());
- // assertTableEquals(uncompressed, gzip);
- //
- // final Table lz4 = ParquetTools.readTable(TestParquetTools.class.getResource("/e2/lz4.parquet").getFile());
- // assertTableEquals(uncompressed, lz4);
- //
- // final Table snappy = ParquetTools.readTable(TestParquetTools.class.getResource("/e2/snappy.parquet").getFile());
- // assertTableEquals(uncompressed, snappy);
- //
- // final Table zstd = ParquetTools.readTable(TestParquetTools.class.getResource("/e2/zstd.parquet").getFile());
- // assertTableEquals(uncompressed, zstd);
- // }
+ @Test
+ public void e2() {
+ final Table uncompressed =
+ ParquetTools.readTable(TestParquetTools.class.getResource("/e2/uncompressed.parquet").getFile());
+
+ final Table gzip = ParquetTools.readTable(TestParquetTools.class.getResource("/e2/gzip.parquet").getFile());
+ assertTableEquals(uncompressed, gzip);
+
+ final Table lz4 = ParquetTools.readTable(TestParquetTools.class.getResource("/e2/lz4.parquet").getFile());
+ assertTableEquals(uncompressed, lz4);
+
+ final Table snappy = ParquetTools.readTable(TestParquetTools.class.getResource("/e2/snappy.parquet").getFile());
+ assertTableEquals(uncompressed, snappy);
+
+ final Table zstd = ParquetTools.readTable(TestParquetTools.class.getResource("/e2/zstd.parquet").getFile());
+ assertTableEquals(uncompressed, zstd);
+ }
 
  private void testWriteRead(Table source, Function<Table, Table> transform) {
  final File f2w = new File(testRoot, "testWriteRead.parquet");

py/server/tests/test_parquet.py

@@ -10,7 +10,7 @@
 import pandas
 import pyarrow.parquet
 
-from deephaven import empty_table, dtypes, new_table
+from deephaven import DHError, empty_table, dtypes, new_table
 from deephaven import arrow as dharrow
 from deephaven.column import InputColumn
 from deephaven.pandas import to_pandas, to_table
@@ -304,7 +304,7 @@ def round_trip_with_compression(self, compression_codec_name, dh_table, vector_c
  # result_table = read('data_from_pandas.parquet')
  # self.assert_table_equals(dh_table, result_table)
 
- def test_writing_via_pyarrow(self):
+ def test_writing_lists_via_pyarrow(self):
  # This function tests that we can write tables with list types to parquet files via pyarrow and read them back
  # through deephaven's parquet reader code with no exceptions
  pa_table = pyarrow.table({'numList': [[2, 2, 4]],
@@ -314,6 +314,25 @@ def test_writing_via_pyarrow(self):
  pa_table_from_disk = dharrow.to_arrow(from_disk)
  self.assertTrue(pa_table.equals(pa_table_from_disk))
 
+ def test_writing_time_via_pyarrow(self):
+ def _test_writing_time_helper(filename):
+ metadata = pyarrow.parquet.read_metadata(filename)
+ if "isAdjustedToUTC=false" in str(metadata.row_group(0).column(0)):
+ # TODO(deephaven-core#976): Unable to read non UTC adjusted timestamps
+ with self.assertRaises(DHError) as e:
+ read(filename)
+ self.assertIn("ParquetFileReaderException", e.exception.root_cause)
+
+ df = pandas.DataFrame({
+ "f": pandas.date_range("20130101", periods=3),
+ })
+ df.to_parquet("pyarrow_26.parquet", engine='pyarrow', compression=None, version='2.6')
+ _test_writing_time_helper("pyarrow_26.parquet")
+ df.to_parquet("pyarrow_24.parquet", engine='pyarrow', compression=None, version='2.4')
+ _test_writing_time_helper("pyarrow_24.parquet")
+ df.to_parquet("pyarrow_10.parquet", engine='pyarrow', compression=None, version='1.0')
+ _test_writing_time_helper("pyarrow_10.parquet")
+
  def test_dictionary_encoding(self):
  dh_table = empty_table(10).update(formulas=[
  "shortStringColumn = `Row ` + i",