You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "talatuyarer (via GitHub)" <gi...@apache.org> on 2025/10/03 09:19:37 UTC
[PR] Add Nanosecond Precision Support for Flink-Iceberg IntegrationFlink nanosecond support [iceberg]
talatuyarer opened a new pull request, #14245:
URL: https://github.com/apache/iceberg/pull/14245
This PR fixes the issue where nanosecond precision timestamps were being truncated to millisecond precision when using Apache Flink with Apache Iceberg V3 tables. Now you can actually use those fancy TIMESTAMP(9) and TIMESTAMP_LTZ(9) types without losing precision!
**The Problem**
When inserting data with nanosecond precision using Flink SQL like:
```sql
INSERT INTO my_table VALUES (TIMESTAMP '2025-01-15 10:30:45.123456789');
```
The data would mysteriously lose precision and come back as:
```
2025-01-15 10:30:45.123456
```
**Core Issues:**
`RowDataWrapper` was converting all timestamps to microseconds regardless of precision
`FlinkParquetWriters` was losing nanosecond precision when writing to Parquet files
`StructRowData` and `RowDataUtil` were truncating precision when reading data back
`FlinkTypeToType` was always mapping `TIMESTAMP(9)` to microsecond Iceberg types (causing Flink to cast to `TIMESTAMP(6)`)
**Data Format Support:**
✅ Parquet: Fixed writers and readers to preserve nanosecond precision
✅ Avro: Already working perfectly (verified with tests)
❌ ORC: Still needs fixing (truncates to microseconds)
If you want to test by yourself.
```sql
-- Create table with nanosecond precision
CREATE TABLE test_table (
id BIGINT,
ts TIMESTAMP(9),
ts_tz TIMESTAMP_LTZ(9)
) WITH (
'connector' = 'iceberg',
'catalog-name' = 'hadoop_catalog',
'warehouse' = 'gs://my-bucket/warehouse',
'format-version' = 3
);
-- Insert with nanosecond precision
INSERT INTO test_table VALUES
(1, TIMESTAMP '2025-01-15 10:30:45.123456789', TIMESTAMP '2025-01-15 10:30:45.123456789');
-- Query and verify precision is preserved
SELECT * FROM test_table;
-- Should show: 2025-01-15 10:30:45.123456789 (not truncated!)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer closed pull request #14245: Add Nanosecond Precision Support for Flink-Iceberg Integration
URL: https://github.com/apache/iceberg/pull/14245
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2626551685
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java:
##########
@@ -137,12 +137,20 @@ public Type visit(TimeType timeType) {
@Override
public Type visit(TimestampType timestampType) {
- return Types.TimestampType.withoutZone();
+ if (timestampType.getPrecision() == 9) {
+ return Types.TimestampNanoType.withoutZone();
+ } else {
+ return Types.TimestampType.withoutZone();
+ }
}
@Override
public Type visit(LocalZonedTimestampType localZonedTimestampType) {
- return Types.TimestampType.withZone();
+ if (localZonedTimestampType.getPrecision() == 9) {
+ return Types.TimestampNanoType.withZone();
+ } else {
+ return Types.TimestampType.withZone();
+ }
Review Comment:
Commenting here again because the other thread has been resolved. We should handle all precisions here for completion, i.e. 7 and 8 as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2405308847
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractorEndToEnd.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.reader;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopTableExtension;
+import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.flink.source.FlinkInputFormat;
+import org.apache.iceberg.flink.source.FlinkSource;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * End-to-end tests for ColumnStatsWatermarkExtractor with nanosecond precision timestamps. This
+ * test validates that watermark extraction works correctly with actual data files containing
+ * nanosecond precision timestamps.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestColumnStatsWatermarkExtractorEndToEnd {
+
+ @Parameter(index = 0)
+ private FileFormat format;
+
+ @Parameters(name = "format = {0}")
+ public static Object[][] parameters() {
+ return new Object[][] {{FileFormat.PARQUET}, {FileFormat.AVRO}};
+ }
+
+ private static final String DATABASE = "test_db";
+ private static final String TABLE = "test_watermark_table";
+
+ // Schema with nanosecond precision timestamps for watermark testing
+ private static final Schema NANOSECOND_WATERMARK_SCHEMA =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "event_time_ns", Types.TimestampNanoType.withoutZone()),
+ required(3, "event_time_ns_tz", Types.TimestampNanoType.withZone()),
+ required(4, "event_time_us", Types.TimestampType.withoutZone()),
+ required(5, "data", Types.StringType.get()));
+
+ @TempDir protected Path temporaryDirectory;
+
+ @RegisterExtension
+ private static final HadoopTableExtension TABLE_EXTENSION =
+ HadoopTableExtension.withFormatVersion3(DATABASE, TABLE, NANOSECOND_WATERMARK_SCHEMA);
+
+ @Test
+ public void testWatermarkExtractorCreationWithNanosecondTimestamps() {
+ Table table = TABLE_EXTENSION.table();
+
+ // Test that we can create watermark extractors for nanosecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorNs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+ assertThat(extractorNs).isNotNull();
+
+ ColumnStatsWatermarkExtractor extractorNsTz =
+ new ColumnStatsWatermarkExtractor(
+ table.schema(), "event_time_ns_tz", TimeUnit.MICROSECONDS);
+ assertThat(extractorNsTz).isNotNull();
+
+ // Test that we can still create extractors for microsecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorUs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_us", TimeUnit.MICROSECONDS);
+ assertThat(extractorUs).isNotNull();
+ }
+
+ @TestTemplate
+ public void testWatermarkExtractionWithRealData() throws Exception {
Review Comment:
What does this test do?
Do we have test which test the actual extraction?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2606177198
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/StructRowData.java:
##########
@@ -186,8 +186,31 @@ private BigDecimal getDecimalInternal(int pos) {
@Override
public TimestampData getTimestamp(int pos, int precision) {
- long timeLong = getLong(pos);
- return TimestampData.fromEpochMillis(timeLong / 1000, (int) (timeLong % 1000) * 1000);
+ if (precision == 9) {
+ // Nanosecond precision: get nanoseconds directly from struct
+ Object longVal = struct.get(pos, Object.class);
+ long nanos;
+ if (longVal instanceof Long) {
+ nanos = (long) longVal;
+ } else if (longVal instanceof OffsetDateTime) {
+ nanos = Duration.between(Instant.EPOCH, (OffsetDateTime) longVal).toNanos();
+ } else if (longVal instanceof LocalDateTime) {
+ nanos =
+ Duration.between(Instant.EPOCH, ((LocalDateTime) longVal).atOffset(ZoneOffset.UTC))
+ .toNanos();
+ } else {
+ throw new IllegalStateException(
+ "Unknown type for nanosecond timestamp field. Type name: "
+ + longVal.getClass().getName());
+ }
+ long mills = Math.floorDiv(nanos, 1_000_000);
+ int leftover = Math.floorMod(nanos, 1_000_000);
+ return TimestampData.fromEpochMillis(mills, leftover);
Review Comment:
Question: Why not something similar to the line below with `/`, `%` and `*` operators?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2622323561
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/RowDataToAvroGenericRecordConverter.java:
##########
@@ -41,30 +48,116 @@
public class RowDataToAvroGenericRecordConverter implements Function<RowData, GenericRecord> {
private final RowDataToAvroConverters.RowDataToAvroConverter converter;
private final Schema avroSchema;
+ private final Set<Integer> timestampNanosFieldIndices;
+ private final RowData.FieldGetter[] fieldGetters;
- private RowDataToAvroGenericRecordConverter(RowType rowType, Schema avroSchema) {
- this.converter = RowDataToAvroConverters.createConverter(rowType);
+ private RowDataToAvroGenericRecordConverter(
+ RowType converterRowType, Schema avroSchema, Set<Integer> timestampNanosFieldIndices) {
+ this.converter = RowDataToAvroConverters.createConverter(converterRowType);
this.avroSchema = avroSchema;
+ this.timestampNanosFieldIndices = timestampNanosFieldIndices;
+
+ // Pre-create field getters only if there are timestamp-nanos fields
+ // (otherwise we take the early return path and never use them)
+ if (!timestampNanosFieldIndices.isEmpty()) {
+ this.fieldGetters = new RowData.FieldGetter[converterRowType.getFieldCount()];
+ for (int i = 0; i < converterRowType.getFieldCount(); i++) {
+ LogicalType fieldType = converterRowType.getTypeAt(i);
+ this.fieldGetters[i] = RowData.createFieldGetter(fieldType, i);
+ }
+ } else {
+ this.fieldGetters = null;
+ }
}
@Override
public GenericRecord apply(RowData rowData) {
- return (GenericRecord) converter.convert(avroSchema, rowData);
+ // Pre-process: Flink's RowDataToAvroConverters expects Long (nanoseconds) for timestamp-nanos,
+ // but our RowData has TimestampData. Convert TimestampData to Long nanoseconds.
+ if (timestampNanosFieldIndices.isEmpty()) {
+ return (GenericRecord) converter.convert(avroSchema, rowData);
+ }
+
+ // Create a new GenericRowData with Long values for timestamp-nanos fields
+ GenericRowData processedRowData = new GenericRowData(rowData.getArity());
+ processedRowData.setRowKind(rowData.getRowKind());
+
+ for (int i = 0; i < rowData.getArity(); i++) {
+ if (timestampNanosFieldIndices.contains(i)) {
+ // Convert TimestampData to Long nanoseconds
+ if (!rowData.isNullAt(i)) {
+ TimestampData timestampData = rowData.getTimestamp(i, 9);
+ long nanos =
+ timestampData.getMillisecond() * 1_000_000L + timestampData.getNanoOfMillisecond();
+ processedRowData.setField(i, nanos);
+ } else {
+ processedRowData.setField(i, null);
+ }
+ } else {
+ // Copy other fields as-is using pre-created field getter
+ processedRowData.setField(i, fieldGetters[i].getFieldOrNull(rowData));
+ }
+ }
+
+ return (GenericRecord) converter.convert(avroSchema, processedRowData);
}
/** Create a converter based on Iceberg schema */
public static RowDataToAvroGenericRecordConverter fromIcebergSchema(
String tableName, org.apache.iceberg.Schema icebergSchema) {
- RowType rowType = FlinkSchemaUtil.convert(icebergSchema);
+ RowType originalRowType = FlinkSchemaUtil.convert(icebergSchema);
Schema avroSchema = AvroSchemaUtil.convert(icebergSchema, tableName);
- return new RowDataToAvroGenericRecordConverter(rowType, avroSchema);
+
+ // Detect timestamp-nanos fields and create converter RowType with BIGINT for those fields
+ List<LogicalType> converterFields = Lists.newArrayList();
+ Set<Integer> timestampNanosFieldIndices = new java.util.HashSet<>();
Review Comment:
NIT:
```suggestion
Set<Integer> timestampNanosFieldIndices = Sets.newHashSet();
```
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java:
##########
@@ -137,12 +137,30 @@ public Type visit(TimeType timeType) {
@Override
public Type visit(TimestampType timestampType) {
- return Types.TimestampType.withoutZone();
+ int precision = timestampType.getPrecision();
+ if (precision > 9) {
+ throw new IllegalArgumentException(
+ "Unsupported timestamp precision: " + precision + ". Maximum supported precision is 9.");
+ }
+ if (precision == 9) {
+ return Types.TimestampNanoType.withoutZone();
+ } else {
Review Comment:
Yes, they are possible. Not if this Flink type has been constructed from an Iceberg table, but there are other ways in Flink to create a timestamp with precision < 9.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3370476121
@talatuyarer: Could we please limit the changes to Flink 2.1, so it is easier to review, and easier to update based on the review comments?
Also, could you please review the generated tests, and make sure that the relevant tests are there and we don't have tests which are redundant?
One mora suggestion: DataGenerators is used in tests to automatically add tests for the types. Please add timestamp nano to them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414574923
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/HadoopTableExtension.java:
##########
@@ -18,42 +18,95 @@
*/
package org.apache.iceberg.flink;
+import java.util.Map;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.extension.ExtensionContext;
public class HadoopTableExtension extends HadoopCatalogExtension {
private final Schema schema;
private final PartitionSpec partitionSpec;
+ private final Map<String, String> properties;
private Table table;
public HadoopTableExtension(String database, String tableName, Schema schema) {
- this(database, tableName, schema, null);
+ this(database, tableName, schema, null, null);
}
public HadoopTableExtension(
String database, String tableName, Schema schema, PartitionSpec partitionSpec) {
+ this(database, tableName, schema, partitionSpec, null);
+ }
+
+ public HadoopTableExtension(
+ String database, String tableName, Schema schema, Map<String, String> properties) {
+ this(database, tableName, schema, null, properties);
+ }
+
+ public HadoopTableExtension(
+ String database,
+ String tableName,
+ Schema schema,
+ PartitionSpec partitionSpec,
+ Map<String, String> properties) {
super(database, tableName);
this.schema = schema;
this.partitionSpec = partitionSpec;
+ this.properties = properties;
}
@Override
public void beforeEach(ExtensionContext context) throws Exception {
super.beforeEach(context);
if (partitionSpec == null) {
- this.table = catalog.createTable(TableIdentifier.of(database, tableName), schema);
+ if (properties == null) {
+ this.table = catalog.createTable(TableIdentifier.of(database, tableName), schema);
+ } else {
+ this.table =
+ catalog.createTable(
+ TableIdentifier.of(database, tableName),
+ schema,
+ PartitionSpec.unpartitioned(),
+ null,
+ properties);
+ }
} else {
- this.table =
- catalog.createTable(TableIdentifier.of(database, tableName), schema, partitionSpec);
+ if (properties == null) {
+ this.table =
+ catalog.createTable(TableIdentifier.of(database, tableName), schema, partitionSpec);
+ } else {
+ this.table =
+ catalog.createTable(
+ TableIdentifier.of(database, tableName), schema, partitionSpec, null, properties);
+ }
}
tableLoader.open();
}
public Table table() {
return table;
}
+
+ /** Create a HadoopTableExtension with format version 3 for nanosecond timestamp support. */
+ public static HadoopTableExtension withFormatVersion3(
+ String database, String tableName, Schema schema) {
+ return new HadoopTableExtension(
+ database, tableName, schema, ImmutableMap.of(TableProperties.FORMAT_VERSION, "3"));
+ }
+
+ /** Create a HadoopTableExtension with format version 3 for nanosecond timestamp support. */
+ public static HadoopTableExtension withFormatVersion3(
Review Comment:
Is this used?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2405264615
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/StructRowData.java:
##########
@@ -186,7 +186,13 @@ private BigDecimal getDecimalInternal(int pos) {
@Override
public TimestampData getTimestamp(int pos, int precision) {
long timeLong = getLong(pos);
- return TimestampData.fromEpochMillis(timeLong / 1000, (int) (timeLong % 1000) * 1000);
+ if (precision == 9) {
+ // For nanosecond precision, preserve the full nanosecond value
+ return TimestampData.fromEpochMillis(timeLong / 1_000_000L, (int) (timeLong % 1_000_000L));
+ } else {
+ // For microsecond precision, use the original logic
+ return TimestampData.fromEpochMillis(timeLong / 1000, (int) (timeLong % 1000) * 1000);
+ }
Review Comment:
Should we handle where precision is different than 9, and 6?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3793114296
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2626551685
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java:
##########
@@ -137,12 +137,20 @@ public Type visit(TimeType timeType) {
@Override
public Type visit(TimestampType timestampType) {
- return Types.TimestampType.withoutZone();
+ if (timestampType.getPrecision() == 9) {
+ return Types.TimestampNanoType.withoutZone();
+ } else {
+ return Types.TimestampType.withoutZone();
+ }
}
@Override
public Type visit(LocalZonedTimestampType localZonedTimestampType) {
- return Types.TimestampType.withZone();
+ if (localZonedTimestampType.getPrecision() == 9) {
+ return Types.TimestampNanoType.withZone();
+ } else {
+ return Types.TimestampType.withZone();
+ }
Review Comment:
Commenting here again because the other thread has been resolved and somehow cannot be unresolved. We should handle all precisions here for completion, i.e. 7 and 8 as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2606202092
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/AvroGenericRecordToRowDataMapper.java:
##########
@@ -18,44 +18,119 @@
*/
package org.apache.iceberg.flink.sink;
+import java.util.List;
+import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.formats.avro.AvroToRowDataConverters;
import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
+import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.utils.TypeConversions;
-import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
/**
* This util class converts Avro GenericRecord to Flink RowData. <br>
* <br>
* Internally it uses Flink {@link AvroToRowDataConverters}. Because of the precision difference
* between how Iceberg schema (micro) and Flink {@link AvroToRowDataConverters} (milli) deal with
- * time type, we can't directly use the Avro Schema converted from Iceberg schema via {@link
- * AvroSchemaUtil#convert(org.apache.iceberg.Schema, String)}.
+ * time type, we can't directly use the Avro Schema converted from Iceberg schema. Additionally,
+ * Flink's converter doesn't handle timestamp-nanos logical type properly, so we manually fix those
+ * fields after conversion.
*/
public class AvroGenericRecordToRowDataMapper implements MapFunction<GenericRecord, RowData> {
private final AvroToRowDataConverters.AvroToRowDataConverter converter;
+ private final List<Integer> timestampNanosFieldIndices;
+ private final RowType rowType;
- AvroGenericRecordToRowDataMapper(RowType rowType) {
+ AvroGenericRecordToRowDataMapper(RowType rowType, List<Integer> timestampNanosFieldIndices) {
this.converter = AvroToRowDataConverters.createRowConverter(rowType);
+ this.timestampNanosFieldIndices = timestampNanosFieldIndices;
+ this.rowType = rowType;
}
@Override
public RowData map(GenericRecord genericRecord) throws Exception {
- return (RowData) converter.convert(genericRecord);
+ RowData rowData = (RowData) converter.convert(genericRecord);
+
+ // Post-process: Flink's AvroToRowDataConverters doesn't properly handle timestamp-nanos,
+ // so we need to manually fix the timestamp fields after conversion
+ if (timestampNanosFieldIndices.isEmpty()) {
+ return rowData;
+ }
+
+ // Create a new GenericRowData with corrected timestamp-nanos fields
+ GenericRowData correctedRowData = new GenericRowData(rowData.getArity());
+ correctedRowData.setRowKind(rowData.getRowKind());
+
+ // Copy all fields from the converted RowData
+ for (int i = 0; i < rowData.getArity(); i++) {
+ if (timestampNanosFieldIndices.contains(i)) {
+ // Manually convert timestamp-nanos field from original Avro record
+ Object avroValue = genericRecord.get(i);
+ if (avroValue instanceof Long) {
+ long nanos = (Long) avroValue;
+ long millis = Math.floorDiv(nanos, 1_000_000);
+ int nanosOfMillis = Math.floorMod(nanos, 1_000_000);
+ TimestampData timestampData = TimestampData.fromEpochMillis(millis, nanosOfMillis);
Review Comment:
Same question as above. Any specific reason to use math?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg IntegrationFlink nanosecond support [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3366128890
@rodmeneses was working on this some time ago. You can get inspiration from the PR ge created: https://github.com/apache/iceberg/pull/11348
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2405309498
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractorEndToEnd.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.reader;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopTableExtension;
+import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.flink.source.FlinkInputFormat;
+import org.apache.iceberg.flink.source.FlinkSource;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * End-to-end tests for ColumnStatsWatermarkExtractor with nanosecond precision timestamps. This
+ * test validates that watermark extraction works correctly with actual data files containing
+ * nanosecond precision timestamps.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestColumnStatsWatermarkExtractorEndToEnd {
+
+ @Parameter(index = 0)
+ private FileFormat format;
+
+ @Parameters(name = "format = {0}")
+ public static Object[][] parameters() {
+ return new Object[][] {{FileFormat.PARQUET}, {FileFormat.AVRO}};
+ }
+
+ private static final String DATABASE = "test_db";
+ private static final String TABLE = "test_watermark_table";
+
+ // Schema with nanosecond precision timestamps for watermark testing
+ private static final Schema NANOSECOND_WATERMARK_SCHEMA =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "event_time_ns", Types.TimestampNanoType.withoutZone()),
+ required(3, "event_time_ns_tz", Types.TimestampNanoType.withZone()),
+ required(4, "event_time_us", Types.TimestampType.withoutZone()),
+ required(5, "data", Types.StringType.get()));
+
+ @TempDir protected Path temporaryDirectory;
+
+ @RegisterExtension
+ private static final HadoopTableExtension TABLE_EXTENSION =
+ HadoopTableExtension.withFormatVersion3(DATABASE, TABLE, NANOSECOND_WATERMARK_SCHEMA);
+
+ @Test
+ public void testWatermarkExtractorCreationWithNanosecondTimestamps() {
+ Table table = TABLE_EXTENSION.table();
+
+ // Test that we can create watermark extractors for nanosecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorNs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+ assertThat(extractorNs).isNotNull();
+
+ ColumnStatsWatermarkExtractor extractorNsTz =
+ new ColumnStatsWatermarkExtractor(
+ table.schema(), "event_time_ns_tz", TimeUnit.MICROSECONDS);
+ assertThat(extractorNsTz).isNotNull();
+
+ // Test that we can still create extractors for microsecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorUs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_us", TimeUnit.MICROSECONDS);
+ assertThat(extractorUs).isNotNull();
+ }
+
+ @TestTemplate
+ public void testWatermarkExtractionWithRealData() throws Exception {
+ Table table = TABLE_EXTENSION.table();
+
+ // Generate test data with nanosecond precision timestamps
+ List<Record> testRecords = RandomGenericData.generate(NANOSECOND_WATERMARK_SCHEMA, 5, 42L);
+
+ // Write data to the table
+ new GenericAppenderHelper(table, format, temporaryDirectory).appendToTable(testRecords);
+
+ // Create watermark extractor for nanosecond timestamp column
+ ColumnStatsWatermarkExtractor extractor =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+
+ // Create a mock split with the actual table data
+ // Note: In a real scenario, this would be created by the Flink source system
+ // For this test, we'll validate that the extractor can be created and configured correctly
+ assertThat(extractor).isNotNull();
+
+ // Verify that the table has the expected schema
+ Schema tableSchema = table.schema();
+ assertThat(tableSchema.findField("event_time_ns")).isNotNull();
+ assertThat(tableSchema.findField("event_time_ns_tz")).isNotNull();
+ assertThat(tableSchema.findField("event_time_us")).isNotNull();
+
+ // Verify that the timestamp fields are of the correct types
+ Types.NestedField eventTimeNsField = tableSchema.findField("event_time_ns");
+ assertThat(eventTimeNsField.type()).isInstanceOf(Types.TimestampNanoType.class);
+
+ Types.NestedField eventTimeNsTzField = tableSchema.findField("event_time_ns_tz");
+ assertThat(eventTimeNsTzField.type()).isInstanceOf(Types.TimestampNanoType.class);
+
+ Types.NestedField eventTimeUsField = tableSchema.findField("event_time_us");
+ assertThat(eventTimeUsField.type()).isInstanceOf(Types.TimestampType.class);
+ }
+
+ @TestTemplate
+ public void testDataScanningWithNanosecondTimestamps() throws Exception {
Review Comment:
What is the goal of this test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2409141783
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractorEndToEnd.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.reader;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopTableExtension;
+import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.flink.source.FlinkInputFormat;
+import org.apache.iceberg.flink.source.FlinkSource;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * End-to-end tests for ColumnStatsWatermarkExtractor with nanosecond precision timestamps. This
+ * test validates that watermark extraction works correctly with actual data files containing
+ * nanosecond precision timestamps.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestColumnStatsWatermarkExtractorEndToEnd {
+
+ @Parameter(index = 0)
+ private FileFormat format;
+
+ @Parameters(name = "format = {0}")
+ public static Object[][] parameters() {
+ return new Object[][] {{FileFormat.PARQUET}, {FileFormat.AVRO}};
+ }
+
+ private static final String DATABASE = "test_db";
+ private static final String TABLE = "test_watermark_table";
+
+ // Schema with nanosecond precision timestamps for watermark testing
+ private static final Schema NANOSECOND_WATERMARK_SCHEMA =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "event_time_ns", Types.TimestampNanoType.withoutZone()),
+ required(3, "event_time_ns_tz", Types.TimestampNanoType.withZone()),
+ required(4, "event_time_us", Types.TimestampType.withoutZone()),
+ required(5, "data", Types.StringType.get()));
+
+ @TempDir protected Path temporaryDirectory;
+
+ @RegisterExtension
+ private static final HadoopTableExtension TABLE_EXTENSION =
+ HadoopTableExtension.withFormatVersion3(DATABASE, TABLE, NANOSECOND_WATERMARK_SCHEMA);
+
+ @Test
+ public void testWatermarkExtractorCreationWithNanosecondTimestamps() {
+ Table table = TABLE_EXTENSION.table();
+
+ // Test that we can create watermark extractors for nanosecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorNs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+ assertThat(extractorNs).isNotNull();
+
+ ColumnStatsWatermarkExtractor extractorNsTz =
+ new ColumnStatsWatermarkExtractor(
+ table.schema(), "event_time_ns_tz", TimeUnit.MICROSECONDS);
+ assertThat(extractorNsTz).isNotNull();
+
+ // Test that we can still create extractors for microsecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorUs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_us", TimeUnit.MICROSECONDS);
+ assertThat(extractorUs).isNotNull();
+ }
+
+ @TestTemplate
+ public void testWatermarkExtractionWithRealData() throws Exception {
+ Table table = TABLE_EXTENSION.table();
+
+ // Generate test data with nanosecond precision timestamps
+ List<Record> testRecords = RandomGenericData.generate(NANOSECOND_WATERMARK_SCHEMA, 5, 42L);
+
+ // Write data to the table
+ new GenericAppenderHelper(table, format, temporaryDirectory).appendToTable(testRecords);
+
+ // Create watermark extractor for nanosecond timestamp column
+ ColumnStatsWatermarkExtractor extractor =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+
+ // Create a mock split with the actual table data
+ // Note: In a real scenario, this would be created by the Flink source system
+ // For this test, we'll validate that the extractor can be created and configured correctly
+ assertThat(extractor).isNotNull();
+
+ // Verify that the table has the expected schema
+ Schema tableSchema = table.schema();
+ assertThat(tableSchema.findField("event_time_ns")).isNotNull();
+ assertThat(tableSchema.findField("event_time_ns_tz")).isNotNull();
+ assertThat(tableSchema.findField("event_time_us")).isNotNull();
+
+ // Verify that the timestamp fields are of the correct types
+ Types.NestedField eventTimeNsField = tableSchema.findField("event_time_ns");
+ assertThat(eventTimeNsField.type()).isInstanceOf(Types.TimestampNanoType.class);
+
+ Types.NestedField eventTimeNsTzField = tableSchema.findField("event_time_ns_tz");
+ assertThat(eventTimeNsTzField.type()).isInstanceOf(Types.TimestampNanoType.class);
+
+ Types.NestedField eventTimeUsField = tableSchema.findField("event_time_us");
+ assertThat(eventTimeUsField.type()).isInstanceOf(Types.TimestampType.class);
+ }
+
+ @TestTemplate
+ public void testDataScanningWithNanosecondTimestamps() throws Exception {
Review Comment:
I added this as Javadoc
```
Tests the full write-read cycle for tables with nanosecond precision timestamps. This verifies that data with nanosecond timestamps can be written to files (Parquet/Avro), the data can be read back through Flink's source implementation, and nanosecond precision is preserved during the round-trip. This is an integration test ensuring that the complete data path works correctly with nanosecond timestamps.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2439476221
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java:
##########
@@ -404,8 +404,10 @@ private static class NanosToTimestampReader
@Override
public TimestampData read(TimestampData ignored) {
long value = readLong();
+ long mills = Math.floorDiv(value, 1_000_000_000L);
+ long nanos = Math.floorMod(value, 1_000_000_000L);
return TimestampData.fromEpochMillis(
- Math.floorDiv(value, 1_000_000L), Math.floorMod(value, 1_000_000));
+ mills + Math.floorDiv(nanos, 1_000_000L), (int) Math.floorMod(nanos, 1_000_000L));
Review Comment:
I don't understand this change.
Why are we precalculating seconds? And then again converting to millis?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3591524990
I think we need to fix this issue. @talatuyarer Could you please let me know the current progress? If there are no further updates, I'd like to continue working on this PR.
Thanks！
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2405285230
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractorEndToEnd.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.reader;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopTableExtension;
+import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.flink.source.FlinkInputFormat;
+import org.apache.iceberg.flink.source.FlinkSource;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * End-to-end tests for ColumnStatsWatermarkExtractor with nanosecond precision timestamps. This
+ * test validates that watermark extraction works correctly with actual data files containing
+ * nanosecond precision timestamps.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestColumnStatsWatermarkExtractorEndToEnd {
+
+ @Parameter(index = 0)
+ private FileFormat format;
+
+ @Parameters(name = "format = {0}")
+ public static Object[][] parameters() {
+ return new Object[][] {{FileFormat.PARQUET}, {FileFormat.AVRO}};
+ }
+
+ private static final String DATABASE = "test_db";
+ private static final String TABLE = "test_watermark_table";
+
+ // Schema with nanosecond precision timestamps for watermark testing
+ private static final Schema NANOSECOND_WATERMARK_SCHEMA =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "event_time_ns", Types.TimestampNanoType.withoutZone()),
+ required(3, "event_time_ns_tz", Types.TimestampNanoType.withZone()),
+ required(4, "event_time_us", Types.TimestampType.withoutZone()),
Review Comment:
question: what does `us` stand for?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3591951789
@pvary and @Guosmilesmile Next week I will update this PR. I am working on this. Sorrty this took longer than I expect.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414431738
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/RowDataToAvroGenericRecordConverter.java:
##########
@@ -41,17 +44,114 @@
public class RowDataToAvroGenericRecordConverter implements Function<RowData, GenericRecord> {
private final RowDataToAvroConverters.RowDataToAvroConverter converter;
private final Schema avroSchema;
+ private final RowType rowType;
private RowDataToAvroGenericRecordConverter(RowType rowType, Schema avroSchema) {
this.converter = RowDataToAvroConverters.createConverter(rowType);
this.avroSchema = avroSchema;
+ this.rowType = rowType;
}
@Override
public GenericRecord apply(RowData rowData) {
+ // Check if schema has timestamp-nanos fields
+ if (needsCustomConversion(avroSchema)) {
+ return convertWithNanos(rowData);
+ }
return (GenericRecord) converter.convert(avroSchema, rowData);
}
+ private boolean needsCustomConversion(Schema schema) {
+ for (Schema.Field field : schema.getFields()) {
+ if (field.schema().getLogicalType() instanceof LogicalTypes.TimestampNanos) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private GenericRecord convertWithNanos(RowData rowData) {
+ GenericRecord result = new GenericData.Record(avroSchema);
+
+ for (int i = 0; i < avroSchema.getFields().size(); i++) {
+ Schema.Field field = avroSchema.getFields().get(i);
+ if (rowData.isNullAt(i)) {
+ result.put(i, null);
+ } else if (field.schema().getLogicalType() instanceof LogicalTypes.TimestampNanos) {
+ // Handle timestamp-nanos by converting TimestampData to long nanos
+ // Get precision from the rowType
+ int precision = 6; // default
+ if (rowType.getTypeAt(i).asSummaryString().contains("(9)")) {
+ precision = 9;
+ }
+ TimestampData timestampData = rowData.getTimestamp(i, precision);
+ long nanos =
+ timestampData.getMillisecond() * 1_000_000L + timestampData.getNanoOfMillisecond();
+ result.put(i, nanos);
+ } else {
+ // For non-timestamp-nanos fields, copy value directly from rowData
+ result.put(i, getFieldValue(rowData, i, field));
+ }
+ }
+ return result;
Review Comment:
Add a newline after closing every block?
See: https://iceberg.apache.org/contribute/#block-spacing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414568067
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractorNanosecond.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.reader;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+
+/** Test that ColumnStatsWatermarkExtractor properly handles nanosecond precision timestamps. */
+public class TestColumnStatsWatermarkExtractorNanosecond {
+
+ @Test
+ public void testNanosecondTimestampTypeDetection() {
+ // Create a schema with nanosecond precision timestamp
+ Schema schema =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "timestamp_ns", Types.TimestampNanoType.withoutZone()),
+ required(3, "timestamp_ns_tz", Types.TimestampNanoType.withZone()),
+ required(4, "timestamp_us", Types.TimestampType.withoutZone()));
+
+ // Test nanosecond timestamp without timezone
+ ColumnStatsWatermarkExtractor extractorNs =
+ new ColumnStatsWatermarkExtractor(schema, "timestamp_ns", TimeUnit.MICROSECONDS);
+
+ // Test nanosecond timestamp with timezone
+ ColumnStatsWatermarkExtractor extractorNsTz =
+ new ColumnStatsWatermarkExtractor(schema, "timestamp_ns_tz", TimeUnit.MICROSECONDS);
+
+ // Test microsecond timestamp
+ ColumnStatsWatermarkExtractor extractorUs =
+ new ColumnStatsWatermarkExtractor(schema, "timestamp_us", TimeUnit.MICROSECONDS);
+
+ // Verify that the extractors are created successfully
+ // The actual time unit detection is tested through the constructor logic
+ assertThat(extractorNs).isNotNull();
+ assertThat(extractorNsTz).isNotNull();
+ assertThat(extractorUs).isNotNull();
+ }
+
+ @Test
+ public void testLongColumnWithNanosecondTimeUnit() {
+ // Create a schema with a Long column
+ Schema schema =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "timestamp_long", Types.LongType.get()));
+
+ // Test Long column with nanosecond time unit
+ ColumnStatsWatermarkExtractor extractor =
+ new ColumnStatsWatermarkExtractor(schema, "timestamp_long", TimeUnit.NANOSECONDS);
+
+ // Verify that the extractor is created successfully
+ assertThat(extractor).isNotNull();
+ }
+
+ @Test
+ public void testTimestampTypeDetection() {
Review Comment:
What does this test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414576140
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/HadoopTableExtension.java:
##########
@@ -18,42 +18,95 @@
*/
package org.apache.iceberg.flink;
+import java.util.Map;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.junit.jupiter.api.extension.ExtensionContext;
public class HadoopTableExtension extends HadoopCatalogExtension {
private final Schema schema;
private final PartitionSpec partitionSpec;
+ private final Map<String, String> properties;
private Table table;
public HadoopTableExtension(String database, String tableName, Schema schema) {
- this(database, tableName, schema, null);
+ this(database, tableName, schema, null, null);
}
public HadoopTableExtension(
String database, String tableName, Schema schema, PartitionSpec partitionSpec) {
+ this(database, tableName, schema, partitionSpec, null);
+ }
+
+ public HadoopTableExtension(
+ String database, String tableName, Schema schema, Map<String, String> properties) {
+ this(database, tableName, schema, null, properties);
+ }
+
+ public HadoopTableExtension(
+ String database,
+ String tableName,
+ Schema schema,
+ PartitionSpec partitionSpec,
+ Map<String, String> properties) {
super(database, tableName);
this.schema = schema;
this.partitionSpec = partitionSpec;
+ this.properties = properties;
}
@Override
public void beforeEach(ExtensionContext context) throws Exception {
super.beforeEach(context);
if (partitionSpec == null) {
- this.table = catalog.createTable(TableIdentifier.of(database, tableName), schema);
+ if (properties == null) {
+ this.table = catalog.createTable(TableIdentifier.of(database, tableName), schema);
+ } else {
+ this.table =
+ catalog.createTable(
+ TableIdentifier.of(database, tableName),
+ schema,
+ PartitionSpec.unpartitioned(),
+ null,
+ properties);
+ }
} else {
- this.table =
- catalog.createTable(TableIdentifier.of(database, tableName), schema, partitionSpec);
+ if (properties == null) {
+ this.table =
+ catalog.createTable(TableIdentifier.of(database, tableName), schema, partitionSpec);
+ } else {
+ this.table =
+ catalog.createTable(
+ TableIdentifier.of(database, tableName), schema, partitionSpec, null, properties);
+ }
}
tableLoader.open();
}
public Table table() {
return table;
}
+
+ /** Create a HadoopTableExtension with format version 3 for nanosecond timestamp support. */
+ public static HadoopTableExtension withFormatVersion3(
Review Comment:
Could we just use the constructor instead of creating a new method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414545825
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/RowDataToAvroGenericRecordConverter.java:
##########
@@ -41,17 +44,114 @@
public class RowDataToAvroGenericRecordConverter implements Function<RowData, GenericRecord> {
private final RowDataToAvroConverters.RowDataToAvroConverter converter;
private final Schema avroSchema;
+ private final RowType rowType;
private RowDataToAvroGenericRecordConverter(RowType rowType, Schema avroSchema) {
this.converter = RowDataToAvroConverters.createConverter(rowType);
this.avroSchema = avroSchema;
+ this.rowType = rowType;
}
@Override
public GenericRecord apply(RowData rowData) {
+ // Check if schema has timestamp-nanos fields
+ if (needsCustomConversion(avroSchema)) {
Review Comment:
I don't really like these custom conversions. I feel they are brittle, and don't handle every case.
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/AvroGenericRecordToRowDataMapper.java:
##########
@@ -41,13 +45,114 @@
public class AvroGenericRecordToRowDataMapper implements MapFunction<GenericRecord, RowData> {
private final AvroToRowDataConverters.AvroToRowDataConverter converter;
+ private final Schema avroSchema;
+ private final Function<GenericRecord, RowData> customConverter;
- AvroGenericRecordToRowDataMapper(RowType rowType) {
+ AvroGenericRecordToRowDataMapper(RowType rowType, Schema avroSchema) {
+ this.avroSchema = avroSchema;
this.converter = AvroToRowDataConverters.createRowConverter(rowType);
+ // Check if we need custom conversion for timestamp-nanos
+ this.customConverter = needsCustomConversion(avroSchema) ? this::convertWithNanos : null;
Review Comment:
I don't really like these custom conversions. I feel they are brittle, and don't handle every case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2606184248
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/AvroGenericRecordToRowDataMapper.java:
##########
@@ -18,44 +18,119 @@
*/
package org.apache.iceberg.flink.sink;
+import java.util.List;
+import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.formats.avro.AvroToRowDataConverters;
import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
+import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.utils.TypeConversions;
-import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
/**
* This util class converts Avro GenericRecord to Flink RowData. <br>
* <br>
* Internally it uses Flink {@link AvroToRowDataConverters}. Because of the precision difference
* between how Iceberg schema (micro) and Flink {@link AvroToRowDataConverters} (milli) deal with
- * time type, we can't directly use the Avro Schema converted from Iceberg schema via {@link
- * AvroSchemaUtil#convert(org.apache.iceberg.Schema, String)}.
+ * time type, we can't directly use the Avro Schema converted from Iceberg schema. Additionally,
+ * Flink's converter doesn't handle timestamp-nanos logical type properly, so we manually fix those
+ * fields after conversion.
*/
public class AvroGenericRecordToRowDataMapper implements MapFunction<GenericRecord, RowData> {
private final AvroToRowDataConverters.AvroToRowDataConverter converter;
+ private final List<Integer> timestampNanosFieldIndices;
+ private final RowType rowType;
- AvroGenericRecordToRowDataMapper(RowType rowType) {
+ AvroGenericRecordToRowDataMapper(RowType rowType, List<Integer> timestampNanosFieldIndices) {
this.converter = AvroToRowDataConverters.createRowConverter(rowType);
+ this.timestampNanosFieldIndices = timestampNanosFieldIndices;
+ this.rowType = rowType;
}
@Override
public RowData map(GenericRecord genericRecord) throws Exception {
- return (RowData) converter.convert(genericRecord);
+ RowData rowData = (RowData) converter.convert(genericRecord);
+
+ // Post-process: Flink's AvroToRowDataConverters doesn't properly handle timestamp-nanos,
+ // so we need to manually fix the timestamp fields after conversion
+ if (timestampNanosFieldIndices.isEmpty()) {
+ return rowData;
+ }
+
+ // Create a new GenericRowData with corrected timestamp-nanos fields
+ GenericRowData correctedRowData = new GenericRowData(rowData.getArity());
+ correctedRowData.setRowKind(rowData.getRowKind());
+
+ // Copy all fields from the converted RowData
+ for (int i = 0; i < rowData.getArity(); i++) {
+ if (timestampNanosFieldIndices.contains(i)) {
+ // Manually convert timestamp-nanos field from original Avro record
+ Object avroValue = genericRecord.get(i);
+ if (avroValue instanceof Long) {
+ long nanos = (Long) avroValue;
+ long millis = Math.floorDiv(nanos, 1_000_000);
+ int nanosOfMillis = Math.floorMod(nanos, 1_000_000);
+ TimestampData timestampData = TimestampData.fromEpochMillis(millis, nanosOfMillis);
+ correctedRowData.setField(i, timestampData);
+ } else {
+ // If not a Long, copy as-is
+ correctedRowData.setField(i, getFieldValue(rowData, i));
+ }
+ } else {
+ // Copy other fields as-is
+ correctedRowData.setField(i, getFieldValue(rowData, i));
+ }
+ }
+
+ return correctedRowData;
+ }
+
+ private Object getFieldValue(RowData rowData, int pos) {
+ if (rowData.isNullAt(pos)) {
+ return null;
+ }
Review Comment:
nit: newline
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414565718
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractorEndToEnd.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.reader;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopTableExtension;
+import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.flink.source.FlinkInputFormat;
+import org.apache.iceberg.flink.source.FlinkSource;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * End-to-end tests for ColumnStatsWatermarkExtractor with nanosecond precision timestamps. This
+ * test validates that watermark extraction works correctly with actual data files containing
+ * nanosecond precision timestamps.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestColumnStatsWatermarkExtractorEndToEnd {
+
+ @Parameter(index = 0)
+ private FileFormat format;
+
+ @Parameters(name = "format = {0}")
+ public static Object[][] parameters() {
+ return new Object[][] {{FileFormat.PARQUET}, {FileFormat.AVRO}};
+ }
+
+ private static final String DATABASE = "test_db";
+ private static final String TABLE = "test_watermark_table";
+
+ // Schema with nanosecond precision timestamps for watermark testing
+ // Contains both nanosecond (ns) and microsecond (micros) precision timestamps
+ private static final Schema NANOSECOND_WATERMARK_SCHEMA =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "event_time_ns", Types.TimestampNanoType.withoutZone()),
+ required(3, "event_time_ns_tz", Types.TimestampNanoType.withZone()),
+ required(4, "event_time_micros", Types.TimestampType.withoutZone()),
+ required(5, "data", Types.StringType.get()));
+
+ @TempDir protected Path temporaryDirectory;
+
+ @RegisterExtension
+ private static final HadoopTableExtension TABLE_EXTENSION =
+ HadoopTableExtension.withFormatVersion3(DATABASE, TABLE, NANOSECOND_WATERMARK_SCHEMA);
+
+ /**
+ * Tests that ColumnStatsWatermarkExtractor can be instantiated with both nanosecond and
+ * microsecond precision timestamp columns. This verifies that the extractor correctly recognizes
+ * and accepts both TIMESTAMP_NANO and TIMESTAMP column types.
+ */
+ @Test
+ public void testWatermarkExtractorCreationWithNanosecondTimestamps() {
+ Table table = TABLE_EXTENSION.table();
+
+ // Test that we can create watermark extractors for nanosecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorNs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+ assertThat(extractorNs).isNotNull();
+
+ ColumnStatsWatermarkExtractor extractorNsTz =
+ new ColumnStatsWatermarkExtractor(
+ table.schema(), "event_time_ns_tz", TimeUnit.MICROSECONDS);
+ assertThat(extractorNsTz).isNotNull();
+
+ // Test that we can still create extractors for microsecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorMicros =
+ new ColumnStatsWatermarkExtractor(
+ table.schema(), "event_time_micros", TimeUnit.MICROSECONDS);
+ assertThat(extractorMicros).isNotNull();
+ }
+
+ /**
+ * Tests that ColumnStatsWatermarkExtractor can be created and used with actual data files
+ * containing nanosecond precision timestamps. This validates the end-to-end integration where
+ * data is written to files (Parquet/Avro) and the extractor is configured to read watermarks from
+ * the column statistics. The test verifies that data can be written with nanosecond timestamps,
+ * watermark extractors can be instantiated for these columns, and the schema correctly identifies
+ * timestamp types (nano vs micro).
+ */
+ @TestTemplate
+ public void testWatermarkExtractionWithRealData() throws Exception {
+ Table table = TABLE_EXTENSION.table();
+
+ // Generate test data with nanosecond precision timestamps
+ List<Record> testRecords = RandomGenericData.generate(NANOSECOND_WATERMARK_SCHEMA, 5, 42L);
+
+ // Write data to the table
+ new GenericAppenderHelper(table, format, temporaryDirectory).appendToTable(testRecords);
+
+ // Create watermark extractor for nanosecond timestamp column
+ ColumnStatsWatermarkExtractor extractor =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+
+ // Validate that the extractor can be created and configured correctly
+ // Note: The actual watermark extraction happens in the Flink source during split processing
+ assertThat(extractor).isNotNull();
+
+ // Verify that the table has the expected schema
+ Schema tableSchema = table.schema();
+ assertThat(tableSchema.findField("event_time_ns")).isNotNull();
+ assertThat(tableSchema.findField("event_time_ns_tz")).isNotNull();
+ assertThat(tableSchema.findField("event_time_micros")).isNotNull();
+
+ // Verify that the timestamp fields are of the correct types
+ Types.NestedField eventTimeNsField = tableSchema.findField("event_time_ns");
+ assertThat(eventTimeNsField.type()).isInstanceOf(Types.TimestampNanoType.class);
+
+ Types.NestedField eventTimeNsTzField = tableSchema.findField("event_time_ns_tz");
+ assertThat(eventTimeNsTzField.type()).isInstanceOf(Types.TimestampNanoType.class);
+
+ Types.NestedField eventTimeMicrosField = tableSchema.findField("event_time_micros");
+ assertThat(eventTimeMicrosField.type()).isInstanceOf(Types.TimestampType.class);
+ }
+
+ /**
+ * Tests the full write-read cycle for tables with nanosecond precision timestamps. This verifies
+ * that data with nanosecond timestamps can be written to files (Parquet/Avro), the data can be
+ * read back through Flink's source implementation, and nanosecond precision is preserved during
+ * the round-trip. This is an integration test ensuring that the complete data path works
+ * correctly with nanosecond timestamps.
+ */
+ @TestTemplate
+ public void testDataScanningWithNanosecondTimestamps() throws Exception {
+ Table table = TABLE_EXTENSION.table();
+
+ // Generate test data
+ List<Record> expectedRecords = RandomGenericData.generate(NANOSECOND_WATERMARK_SCHEMA, 3, 123L);
+
+ // Write data to the table
+ new GenericAppenderHelper(table, format, temporaryDirectory).appendToTable(expectedRecords);
+
+ // Test that we can scan the data back
+ FlinkSource.Builder builder =
+ FlinkSource.forRowData().table(table).tableLoader(TABLE_EXTENSION.tableLoader());
+
+ List<Row> actualRows = runFormat(builder.buildFormat());
+
+ // Verify that we can read the data back correctly
+ assertThat(actualRows).hasSize(expectedRecords.size());
+
+ // The data should be readable, indicating that nanosecond timestamps are properly handled
+ // in the read path as well
+ TestHelpers.assertRecords(actualRows, expectedRecords, NANOSECOND_WATERMARK_SCHEMA);
Review Comment:
Is there a way to test that the correct Watermark is emitted?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3642947849
@mxm, @Guosmilesmile: Could you please review?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2405259265
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java:
##########
@@ -446,8 +446,9 @@ private TimestampDataWriter(ColumnDescriptor desc) {
@Override
public void write(int repetitionLevel, TimestampData value) {
- column.writeLong(
- repetitionLevel, value.getMillisecond() * 1000 + value.getNanoOfMillisecond() / 1000);
+ // For microsecond precision timestamps, convert to microseconds
+ long micros = value.getMillisecond() * 1_000L + value.getNanoOfMillisecond() / 1_000L;
+ column.writeLong(repetitionLevel, micros);
Review Comment:
Why is this change?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3370050297
@pvary I updated PR and fix all issues. I also check @rodmeneses 's PR and included watermark extraction changes to my PR. I implemented all possible places for nanosecond support. I believe this time Flink's nanosecond support is complete. It is ready for your review 😄
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2409133898
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java:
##########
@@ -137,12 +137,20 @@ public Type visit(TimeType timeType) {
@Override
public Type visit(TimestampType timestampType) {
- return Types.TimestampType.withoutZone();
+ if (timestampType.getPrecision() == 9) {
+ return Types.TimestampNanoType.withoutZone();
+ } else {
+ return Types.TimestampType.withoutZone();
+ }
Review Comment:
No This will be behaviour change. Today Flink Iceberg support Timestamp 0 -> 6 and all maps to microsecond. But this is good catch. We should throw exception when someone enter precision 10
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3374921542
Thank you @pvary for reviewing this. As you suggested I removed all other version except 2.1 and also addressed your comments
- Throw unsupported for Timestamp Precision if it is bigger than 9
- Added Timestamp Nano to DataGenerators
- Improved Test Documentation
Please let me know if you see anything missing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3762310309
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3938331140
Yes Sorry for delaying. Let me update PR this week @pvary
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #14245: Add Nanosecond Precision Support for Flink-Iceberg Integration
URL: https://github.com/apache/iceberg/pull/14245
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2405311451
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractorEndToEnd.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.reader;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopTableExtension;
+import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.flink.source.FlinkInputFormat;
+import org.apache.iceberg.flink.source.FlinkSource;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * End-to-end tests for ColumnStatsWatermarkExtractor with nanosecond precision timestamps. This
+ * test validates that watermark extraction works correctly with actual data files containing
+ * nanosecond precision timestamps.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestColumnStatsWatermarkExtractorEndToEnd {
+
+ @Parameter(index = 0)
+ private FileFormat format;
+
+ @Parameters(name = "format = {0}")
+ public static Object[][] parameters() {
+ return new Object[][] {{FileFormat.PARQUET}, {FileFormat.AVRO}};
+ }
+
+ private static final String DATABASE = "test_db";
+ private static final String TABLE = "test_watermark_table";
+
+ // Schema with nanosecond precision timestamps for watermark testing
+ private static final Schema NANOSECOND_WATERMARK_SCHEMA =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "event_time_ns", Types.TimestampNanoType.withoutZone()),
+ required(3, "event_time_ns_tz", Types.TimestampNanoType.withZone()),
+ required(4, "event_time_us", Types.TimestampType.withoutZone()),
+ required(5, "data", Types.StringType.get()));
+
+ @TempDir protected Path temporaryDirectory;
+
+ @RegisterExtension
+ private static final HadoopTableExtension TABLE_EXTENSION =
+ HadoopTableExtension.withFormatVersion3(DATABASE, TABLE, NANOSECOND_WATERMARK_SCHEMA);
+
+ @Test
+ public void testWatermarkExtractorCreationWithNanosecondTimestamps() {
+ Table table = TABLE_EXTENSION.table();
+
+ // Test that we can create watermark extractors for nanosecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorNs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+ assertThat(extractorNs).isNotNull();
+
+ ColumnStatsWatermarkExtractor extractorNsTz =
+ new ColumnStatsWatermarkExtractor(
+ table.schema(), "event_time_ns_tz", TimeUnit.MICROSECONDS);
+ assertThat(extractorNsTz).isNotNull();
+
+ // Test that we can still create extractors for microsecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorUs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_us", TimeUnit.MICROSECONDS);
+ assertThat(extractorUs).isNotNull();
+ }
+
+ @TestTemplate
+ public void testWatermarkExtractionWithRealData() throws Exception {
+ Table table = TABLE_EXTENSION.table();
+
+ // Generate test data with nanosecond precision timestamps
+ List<Record> testRecords = RandomGenericData.generate(NANOSECOND_WATERMARK_SCHEMA, 5, 42L);
+
+ // Write data to the table
+ new GenericAppenderHelper(table, format, temporaryDirectory).appendToTable(testRecords);
+
+ // Create watermark extractor for nanosecond timestamp column
+ ColumnStatsWatermarkExtractor extractor =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+
+ // Create a mock split with the actual table data
+ // Note: In a real scenario, this would be created by the Flink source system
+ // For this test, we'll validate that the extractor can be created and configured correctly
+ assertThat(extractor).isNotNull();
+
+ // Verify that the table has the expected schema
+ Schema tableSchema = table.schema();
+ assertThat(tableSchema.findField("event_time_ns")).isNotNull();
+ assertThat(tableSchema.findField("event_time_ns_tz")).isNotNull();
+ assertThat(tableSchema.findField("event_time_us")).isNotNull();
+
+ // Verify that the timestamp fields are of the correct types
+ Types.NestedField eventTimeNsField = tableSchema.findField("event_time_ns");
+ assertThat(eventTimeNsField.type()).isInstanceOf(Types.TimestampNanoType.class);
+
+ Types.NestedField eventTimeNsTzField = tableSchema.findField("event_time_ns_tz");
+ assertThat(eventTimeNsTzField.type()).isInstanceOf(Types.TimestampNanoType.class);
+
+ Types.NestedField eventTimeUsField = tableSchema.findField("event_time_us");
+ assertThat(eventTimeUsField.type()).isInstanceOf(Types.TimestampType.class);
+ }
+
+ @TestTemplate
+ public void testDataScanningWithNanosecondTimestamps() throws Exception {
+ Table table = TABLE_EXTENSION.table();
+
+ // Generate test data
+ List<Record> expectedRecords = RandomGenericData.generate(NANOSECOND_WATERMARK_SCHEMA, 3, 123L);
+
+ // Write data to the table
+ new GenericAppenderHelper(table, format, temporaryDirectory).appendToTable(expectedRecords);
+
+ // Test that we can scan the data back
+ FlinkSource.Builder builder =
+ FlinkSource.forRowData().table(table).tableLoader(TABLE_EXTENSION.tableLoader());
+
+ List<Row> actualRows = runFormat(builder.buildFormat());
+
+ // Verify that we can read the data back correctly
+ assertThat(actualRows).hasSize(expectedRecords.size());
+
+ // The data should be readable, indicating that nanosecond timestamps are properly handled
+ // in the read path as well
+ TestHelpers.assertRecords(actualRows, expectedRecords, NANOSECOND_WATERMARK_SCHEMA);
+ }
+
+ @Test
+ public void testWatermarkExtractorWithLongColumn() {
Review Comment:
Do we have test which actually test the extraction?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414563442
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractorEndToEnd.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.reader;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopTableExtension;
+import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.flink.source.FlinkInputFormat;
+import org.apache.iceberg.flink.source.FlinkSource;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * End-to-end tests for ColumnStatsWatermarkExtractor with nanosecond precision timestamps. This
+ * test validates that watermark extraction works correctly with actual data files containing
+ * nanosecond precision timestamps.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestColumnStatsWatermarkExtractorEndToEnd {
+
+ @Parameter(index = 0)
+ private FileFormat format;
+
+ @Parameters(name = "format = {0}")
+ public static Object[][] parameters() {
+ return new Object[][] {{FileFormat.PARQUET}, {FileFormat.AVRO}};
+ }
+
+ private static final String DATABASE = "test_db";
+ private static final String TABLE = "test_watermark_table";
+
+ // Schema with nanosecond precision timestamps for watermark testing
+ // Contains both nanosecond (ns) and microsecond (micros) precision timestamps
+ private static final Schema NANOSECOND_WATERMARK_SCHEMA =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "event_time_ns", Types.TimestampNanoType.withoutZone()),
+ required(3, "event_time_ns_tz", Types.TimestampNanoType.withZone()),
+ required(4, "event_time_micros", Types.TimestampType.withoutZone()),
+ required(5, "data", Types.StringType.get()));
+
+ @TempDir protected Path temporaryDirectory;
+
+ @RegisterExtension
+ private static final HadoopTableExtension TABLE_EXTENSION =
+ HadoopTableExtension.withFormatVersion3(DATABASE, TABLE, NANOSECOND_WATERMARK_SCHEMA);
+
+ /**
+ * Tests that ColumnStatsWatermarkExtractor can be instantiated with both nanosecond and
+ * microsecond precision timestamp columns. This verifies that the extractor correctly recognizes
+ * and accepts both TIMESTAMP_NANO and TIMESTAMP column types.
+ */
+ @Test
+ public void testWatermarkExtractorCreationWithNanosecondTimestamps() {
+ Table table = TABLE_EXTENSION.table();
+
+ // Test that we can create watermark extractors for nanosecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorNs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+ assertThat(extractorNs).isNotNull();
+
+ ColumnStatsWatermarkExtractor extractorNsTz =
+ new ColumnStatsWatermarkExtractor(
+ table.schema(), "event_time_ns_tz", TimeUnit.MICROSECONDS);
+ assertThat(extractorNsTz).isNotNull();
+
+ // Test that we can still create extractors for microsecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorMicros =
+ new ColumnStatsWatermarkExtractor(
+ table.schema(), "event_time_micros", TimeUnit.MICROSECONDS);
+ assertThat(extractorMicros).isNotNull();
+ }
+
+ /**
+ * Tests that ColumnStatsWatermarkExtractor can be created and used with actual data files
+ * containing nanosecond precision timestamps. This validates the end-to-end integration where
+ * data is written to files (Parquet/Avro) and the extractor is configured to read watermarks from
+ * the column statistics. The test verifies that data can be written with nanosecond timestamps,
+ * watermark extractors can be instantiated for these columns, and the schema correctly identifies
+ * timestamp types (nano vs micro).
+ */
+ @TestTemplate
+ public void testWatermarkExtractionWithRealData() throws Exception {
+ Table table = TABLE_EXTENSION.table();
+
+ // Generate test data with nanosecond precision timestamps
+ List<Record> testRecords = RandomGenericData.generate(NANOSECOND_WATERMARK_SCHEMA, 5, 42L);
+
+ // Write data to the table
+ new GenericAppenderHelper(table, format, temporaryDirectory).appendToTable(testRecords);
+
+ // Create watermark extractor for nanosecond timestamp column
+ ColumnStatsWatermarkExtractor extractor =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+
+ // Validate that the extractor can be created and configured correctly
+ // Note: The actual watermark extraction happens in the Flink source during split processing
+ assertThat(extractor).isNotNull();
+
+ // Verify that the table has the expected schema
+ Schema tableSchema = table.schema();
+ assertThat(tableSchema.findField("event_time_ns")).isNotNull();
+ assertThat(tableSchema.findField("event_time_ns_tz")).isNotNull();
+ assertThat(tableSchema.findField("event_time_micros")).isNotNull();
Review Comment:
How could this be different?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414429141
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/RowDataToAvroGenericRecordConverter.java:
##########
@@ -41,17 +44,114 @@
public class RowDataToAvroGenericRecordConverter implements Function<RowData, GenericRecord> {
private final RowDataToAvroConverters.RowDataToAvroConverter converter;
private final Schema avroSchema;
+ private final RowType rowType;
private RowDataToAvroGenericRecordConverter(RowType rowType, Schema avroSchema) {
this.converter = RowDataToAvroConverters.createConverter(rowType);
this.avroSchema = avroSchema;
+ this.rowType = rowType;
}
@Override
public GenericRecord apply(RowData rowData) {
+ // Check if schema has timestamp-nanos fields
+ if (needsCustomConversion(avroSchema)) {
+ return convertWithNanos(rowData);
+ }
return (GenericRecord) converter.convert(avroSchema, rowData);
}
+ private boolean needsCustomConversion(Schema schema) {
+ for (Schema.Field field : schema.getFields()) {
+ if (field.schema().getLogicalType() instanceof LogicalTypes.TimestampNanos) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private GenericRecord convertWithNanos(RowData rowData) {
+ GenericRecord result = new GenericData.Record(avroSchema);
+
+ for (int i = 0; i < avroSchema.getFields().size(); i++) {
+ Schema.Field field = avroSchema.getFields().get(i);
+ if (rowData.isNullAt(i)) {
+ result.put(i, null);
+ } else if (field.schema().getLogicalType() instanceof LogicalTypes.TimestampNanos) {
+ // Handle timestamp-nanos by converting TimestampData to long nanos
+ // Get precision from the rowType
+ int precision = 6; // default
+ if (rowType.getTypeAt(i).asSummaryString().contains("(9)")) {
+ precision = 9;
+ }
+ TimestampData timestampData = rowData.getTimestamp(i, precision);
+ long nanos =
+ timestampData.getMillisecond() * 1_000_000L + timestampData.getNanoOfMillisecond();
Review Comment:
Why not do this in the `getFieldValue`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414391808
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/StructRowData.java:
##########
@@ -186,8 +186,34 @@ private BigDecimal getDecimalInternal(int pos) {
@Override
public TimestampData getTimestamp(int pos, int precision) {
- long timeLong = getLong(pos);
- return TimestampData.fromEpochMillis(timeLong / 1000, (int) (timeLong % 1000) * 1000);
+ Object timestampVal = struct.get(pos, Object.class);
+
+ // Convert all timestamp types to nanoseconds (smallest unit)
+ long nanos;
+ if (timestampVal instanceof Long) {
+ // Long values: nanoseconds for timestamp_ns (precision 9), microseconds otherwise
+ nanos = precision == 9 ? (Long) timestampVal : ((Long) timestampVal) * 1000;
+ } else if (timestampVal instanceof OffsetDateTime) {
+ nanos = Duration.between(Instant.EPOCH, (OffsetDateTime) timestampVal).toNanos();
+ } else if (timestampVal instanceof LocalDateTime) {
+ nanos =
+ Duration.between(Instant.EPOCH, ((LocalDateTime) timestampVal).atOffset(ZoneOffset.UTC))
+ .toNanos();
+ } else {
Review Comment:
Is it intentional that we left out `LocalDate` and `LocalTime`? - They were implemented in the `getLong` method
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2405269223
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java:
##########
@@ -137,12 +137,20 @@ public Type visit(TimeType timeType) {
@Override
public Type visit(TimestampType timestampType) {
- return Types.TimestampType.withoutZone();
+ if (timestampType.getPrecision() == 9) {
+ return Types.TimestampNanoType.withoutZone();
+ } else {
+ return Types.TimestampType.withoutZone();
+ }
Review Comment:
Should we throw an exception when the precision is not 9 or 6?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3540598378
What is the status of this one @talatuyarer?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2609607212
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/StructRowData.java:
##########
@@ -186,8 +186,31 @@ private BigDecimal getDecimalInternal(int pos) {
@Override
public TimestampData getTimestamp(int pos, int precision) {
- long timeLong = getLong(pos);
- return TimestampData.fromEpochMillis(timeLong / 1000, (int) (timeLong % 1000) * 1000);
+ if (precision == 9) {
+ // Nanosecond precision: get nanoseconds directly from struct
+ Object longVal = struct.get(pos, Object.class);
+ long nanos;
+ if (longVal instanceof Long) {
+ nanos = (long) longVal;
+ } else if (longVal instanceof OffsetDateTime) {
+ nanos = Duration.between(Instant.EPOCH, (OffsetDateTime) longVal).toNanos();
+ } else if (longVal instanceof LocalDateTime) {
+ nanos =
+ Duration.between(Instant.EPOCH, ((LocalDateTime) longVal).atOffset(ZoneOffset.UTC))
+ .toNanos();
+ } else {
+ throw new IllegalStateException(
+ "Unknown type for nanosecond timestamp field. Type name: "
+ + longVal.getClass().getName());
+ }
+ long mills = Math.floorDiv(nanos, 1_000_000);
+ int leftover = Math.floorMod(nanos, 1_000_000);
+ return TimestampData.fromEpochMillis(mills, leftover);
Review Comment:
Aligned with previous calculation style
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2611377924
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/StructRowData.java:
##########
@@ -186,8 +186,31 @@ private BigDecimal getDecimalInternal(int pos) {
@Override
public TimestampData getTimestamp(int pos, int precision) {
- long timeLong = getLong(pos);
- return TimestampData.fromEpochMillis(timeLong / 1000, (int) (timeLong % 1000) * 1000);
+ if (precision == 9) {
+ // Nanosecond precision: get nanoseconds directly from struct
+ Object longVal = struct.get(pos, Object.class);
+ long nanos;
+ if (longVal instanceof Long) {
+ nanos = (long) longVal;
+ } else if (longVal instanceof OffsetDateTime) {
+ nanos = Duration.between(Instant.EPOCH, (OffsetDateTime) longVal).toNanos();
+ } else if (longVal instanceof LocalDateTime) {
+ nanos =
+ Duration.between(Instant.EPOCH, ((LocalDateTime) longVal).atOffset(ZoneOffset.UTC))
+ .toNanos();
+ } else {
Review Comment:
We do it here:
https://github.com/apache/iceberg/blob/23dc32e5f1130c9d79d71cbf0be5b85162d8976c/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/StructRowData.java#L188-L191
And this is the code behind:
https://github.com/apache/iceberg/blob/23dc32e5f1130c9d79d71cbf0be5b85162d8976c/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/StructRowData.java#L142-L145
TBH, the `LocalDate` and the `LocalTime` handling doesn't seem correct to me.
We might want to skip this for now, and might fix this later on in a separate PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3539522041
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2439457949
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkValueReaders.java:
##########
@@ -191,10 +191,10 @@ private static class TimestampNanosReader implements ValueReader<TimestampData>
@Override
public TimestampData read(Decoder decoder, Object reuse) throws IOException {
- long nanos = decoder.readLong();
- long mills = Math.floorDiv(nanos, 1_000_000);
- int leftover = Math.floorMod(nanos, 1_000_000);
- return TimestampData.fromEpochMillis(mills, leftover);
+ long storedValue = decoder.readLong();
+ long mills = Math.floorDiv(storedValue, 1_000_000L);
+ int leftoverNanos = (int) Math.floorMod(storedValue, 1_000_000L);
+ return TimestampData.fromEpochMillis(mills, leftoverNanos);
Review Comment:
Any relevant change here?
For me this seems like we renamed some of the variables, but nothing has changed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2609423946
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/StructRowData.java:
##########
@@ -186,8 +186,31 @@ private BigDecimal getDecimalInternal(int pos) {
@Override
public TimestampData getTimestamp(int pos, int precision) {
- long timeLong = getLong(pos);
- return TimestampData.fromEpochMillis(timeLong / 1000, (int) (timeLong % 1000) * 1000);
+ if (precision == 9) {
+ // Nanosecond precision: get nanoseconds directly from struct
+ Object longVal = struct.get(pos, Object.class);
+ long nanos;
+ if (longVal instanceof Long) {
+ nanos = (long) longVal;
+ } else if (longVal instanceof OffsetDateTime) {
+ nanos = Duration.between(Instant.EPOCH, (OffsetDateTime) longVal).toNanos();
+ } else if (longVal instanceof LocalDateTime) {
+ nanos =
+ Duration.between(Instant.EPOCH, ((LocalDateTime) longVal).atOffset(ZoneOffset.UTC))
+ .toNanos();
+ } else {
Review Comment:
My understanding LocalDate and LocalTime should not handle in timestamp-nanos fields because they don't represent complete timestamps. Please let me know if I need to handle.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2606236791
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/RowDataToAvroGenericRecordConverter.java:
##########
@@ -41,30 +48,111 @@
public class RowDataToAvroGenericRecordConverter implements Function<RowData, GenericRecord> {
private final RowDataToAvroConverters.RowDataToAvroConverter converter;
private final Schema avroSchema;
+ private final Set<Integer> timestampNanosFieldIndices;
+ private final RowType converterRowType;
- private RowDataToAvroGenericRecordConverter(RowType rowType, Schema avroSchema) {
- this.converter = RowDataToAvroConverters.createConverter(rowType);
+ private RowDataToAvroGenericRecordConverter(
+ RowType converterRowType, Schema avroSchema, Set<Integer> timestampNanosFieldIndices) {
+ this.converter = RowDataToAvroConverters.createConverter(converterRowType);
this.avroSchema = avroSchema;
+ this.timestampNanosFieldIndices = timestampNanosFieldIndices;
+ this.converterRowType = converterRowType;
}
@Override
public GenericRecord apply(RowData rowData) {
- return (GenericRecord) converter.convert(avroSchema, rowData);
+ // Pre-process: Flink's RowDataToAvroConverters expects Long (nanoseconds) for timestamp-nanos,
+ // but our RowData has TimestampData. Convert TimestampData to Long nanoseconds.
+ if (timestampNanosFieldIndices.isEmpty()) {
+ return (GenericRecord) converter.convert(avroSchema, rowData);
+ }
+
+ // Create a new GenericRowData with Long values for timestamp-nanos fields
+ GenericRowData processedRowData = new GenericRowData(rowData.getArity());
+ processedRowData.setRowKind(rowData.getRowKind());
+
+ for (int i = 0; i < rowData.getArity(); i++) {
+ if (timestampNanosFieldIndices.contains(i)) {
+ // Convert TimestampData to Long nanoseconds
+ if (!rowData.isNullAt(i)) {
+ TimestampData timestampData = rowData.getTimestamp(i, 9);
+ long nanos =
+ timestampData.getMillisecond() * 1_000_000L + timestampData.getNanoOfMillisecond();
+ processedRowData.setField(i, nanos);
+ } else {
+ processedRowData.setField(i, null);
+ }
+ } else {
+ // Copy other fields as-is
+ if (rowData.isNullAt(i)) {
+ processedRowData.setField(i, null);
+ } else {
+ LogicalType fieldType = converterRowType.getTypeAt(i);
+ RowData.FieldGetter fieldGetter = RowData.createFieldGetter(fieldType, i);
+ processedRowData.setField(i, fieldGetter.getFieldOrNull(rowData));
Review Comment:
Don't create the fieldGetters every time
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3934966049
@talatuyarer: Any plans to move forward with this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2405264615
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/StructRowData.java:
##########
@@ -186,7 +186,13 @@ private BigDecimal getDecimalInternal(int pos) {
@Override
public TimestampData getTimestamp(int pos, int precision) {
long timeLong = getLong(pos);
- return TimestampData.fromEpochMillis(timeLong / 1000, (int) (timeLong % 1000) * 1000);
+ if (precision == 9) {
+ // For nanosecond precision, preserve the full nanosecond value
+ return TimestampData.fromEpochMillis(timeLong / 1_000_000L, (int) (timeLong % 1_000_000L));
+ } else {
+ // For microsecond precision, use the original logic
+ return TimestampData.fromEpochMillis(timeLong / 1000, (int) (timeLong % 1000) * 1000);
+ }
Review Comment:
Should we handle where precision is different than 9, and 3?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414401619
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/AvroGenericRecordToRowDataMapper.java:
##########
@@ -41,13 +45,114 @@
public class AvroGenericRecordToRowDataMapper implements MapFunction<GenericRecord, RowData> {
private final AvroToRowDataConverters.AvroToRowDataConverter converter;
+ private final Schema avroSchema;
+ private final Function<GenericRecord, RowData> customConverter;
- AvroGenericRecordToRowDataMapper(RowType rowType) {
+ AvroGenericRecordToRowDataMapper(RowType rowType, Schema avroSchema) {
+ this.avroSchema = avroSchema;
this.converter = AvroToRowDataConverters.createRowConverter(rowType);
+ // Check if we need custom conversion for timestamp-nanos
+ this.customConverter = needsCustomConversion(avroSchema) ? this::convertWithNanos : null;
+ }
+
+ private boolean needsCustomConversion(Schema schema) {
+ for (Schema.Field field : schema.getFields()) {
+ if (field.schema().getLogicalType() instanceof LogicalTypes.TimestampNanos) {
+ return true;
+ }
+ }
+ return false;
+ }
Review Comment:
Maybe:
```
return schema.getFields().stream()
.anyMatch(field -> field.schema().getLogicalType() instanceof LogicalTypes.TimestampNanos);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414588757
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestRowDataWrapper.java:
##########
@@ -59,6 +63,57 @@ public void testTime() {
});
}
+ /** Test that nanosecond precision timestamps are preserved correctly. */
+ @Test
+ public void testNanosecondTimestampPrecision() {
+ // Create a specific timestamp with nanosecond precision
+ LocalDateTime testTime = LocalDateTime.of(2025, 10, 2, 10, 15, 30, 123456789);
+ long expectedMicros = DateTimeUtil.microsFromTimestamp(testTime);
+ long expectedNanos = DateTimeUtil.nanosFromTimestamp(testTime);
+
+ // Microsecond precision schema (should use microsFromTimestamp)
+ Schema microsSchema =
+ new Schema(
+ Types.NestedField.required(1, "timestamp_micros", Types.TimestampType.withoutZone()));
+
+ // Nanosecond precision schema (should use nanosFromTimestamp)
+ Schema nanosSchema =
+ new Schema(
+ Types.NestedField.required(
+ 1, "timestamp_nanos", Types.TimestampNanoType.withoutZone()));
+
+ // Create wrappers for both schemas
+ RowDataWrapper microsWrapper =
+ new RowDataWrapper(FlinkSchemaUtil.convert(microsSchema), microsSchema.asStruct());
+ RowDataWrapper nanosWrapper =
+ new RowDataWrapper(FlinkSchemaUtil.convert(nanosSchema), nanosSchema.asStruct());
+
+ // Test with the same seed to get comparable data
+ RowData microsRowData = RandomRowData.generate(microsSchema, 1, 42L).iterator().next();
+ RowData nanosRowData = RandomRowData.generate(nanosSchema, 1, 42L).iterator().next();
Review Comment:
Could we do this with specific values?
The assertions are hard to reason about with "Should be close"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414580112
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestRowDataWrapper.java:
##########
@@ -59,6 +63,57 @@ public void testTime() {
});
}
+ /** Test that nanosecond precision timestamps are preserved correctly. */
+ @Test
+ public void testNanosecondTimestampPrecision() {
+ // Create a specific timestamp with nanosecond precision
+ LocalDateTime testTime = LocalDateTime.of(2025, 10, 2, 10, 15, 30, 123456789);
+ long expectedMicros = DateTimeUtil.microsFromTimestamp(testTime);
+ long expectedNanos = DateTimeUtil.nanosFromTimestamp(testTime);
Review Comment:
Are these used?
##########
gradle.properties:
##########
@@ -32,4 +32,4 @@ org.gradle.parallel=true
org.gradle.configureondemand=true
# explicitly disable the configuration cache
org.gradle.configuration-cache=false
-org.gradle.jvmargs=-Xmx1536m
+org.gradle.jvmargs=-Xmx1536m
Review Comment:
nit: revert
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414538991
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/AvroGenericRecordToRowDataMapper.java:
##########
@@ -41,13 +45,114 @@
public class AvroGenericRecordToRowDataMapper implements MapFunction<GenericRecord, RowData> {
private final AvroToRowDataConverters.AvroToRowDataConverter converter;
+ private final Schema avroSchema;
+ private final Function<GenericRecord, RowData> customConverter;
- AvroGenericRecordToRowDataMapper(RowType rowType) {
+ AvroGenericRecordToRowDataMapper(RowType rowType, Schema avroSchema) {
+ this.avroSchema = avroSchema;
this.converter = AvroToRowDataConverters.createRowConverter(rowType);
+ // Check if we need custom conversion for timestamp-nanos
+ this.customConverter = needsCustomConversion(avroSchema) ? this::convertWithNanos : null;
+ }
+
+ private boolean needsCustomConversion(Schema schema) {
+ for (Schema.Field field : schema.getFields()) {
+ if (field.schema().getLogicalType() instanceof LogicalTypes.TimestampNanos) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private RowData convertWithNanos(GenericRecord genericRecord) {
+ RowData baseRowData = (RowData) converter.convert(genericRecord);
+ GenericRowData result = new GenericRowData(baseRowData.getArity());
+ result.setRowKind(baseRowData.getRowKind());
+
+ for (int i = 0; i < avroSchema.getFields().size(); i++) {
+ Schema.Field field = avroSchema.getFields().get(i);
+ if (field.schema().getLogicalType() instanceof LogicalTypes.TimestampNanos) {
+ // Handle timestamp-nanos by converting from long nanos to TimestampData
+ Long nanos = (Long) genericRecord.get(i);
+ if (nanos == null) {
+ result.setField(i, null);
+ } else {
+ long millis = Math.floorDiv(nanos, 1_000_000L);
+ int nanoOfMillis = (int) Math.floorMod(nanos, 1_000_000L);
+ result.setField(i, TimestampData.fromEpochMillis(millis, nanoOfMillis));
+ }
+ } else {
+ // Copy from base conversion
+ if (baseRowData.isNullAt(i)) {
+ result.setField(i, null);
+ } else {
+ // Get the value from baseRowData using the appropriate getter
+ result.setField(i, getFieldValue(baseRowData, i, field));
+ }
+ }
+ }
+ return result;
+ }
+
+ private Object getFieldValue(RowData rowData, int pos, Schema.Field field) {
+ // This is a simplified version - you may need to handle more types
+ Schema fieldSchema = field.schema();
+ if (fieldSchema.getType() == Schema.Type.UNION) {
+ // Handle nullable fields
+ for (Schema unionSchema : fieldSchema.getTypes()) {
+ if (unionSchema.getType() != Schema.Type.NULL) {
+ fieldSchema = unionSchema;
+ break;
+ }
+ }
+ }
+
+ // Check for logical types first
+ if (fieldSchema.getLogicalType() != null) {
+ if (fieldSchema.getLogicalType() instanceof LogicalTypes.TimestampMillis
+ || fieldSchema.getLogicalType() instanceof LogicalTypes.TimestampMicros) {
+ return rowData.getTimestamp(pos, 6);
+ } else if (fieldSchema.getLogicalType() instanceof LogicalTypes.Decimal) {
+ LogicalTypes.Decimal decimal = (LogicalTypes.Decimal) fieldSchema.getLogicalType();
+ return rowData.getDecimal(pos, decimal.getPrecision(), decimal.getScale());
+ }
+ // timestamp-nanos is handled separately in convertWithNanos
+ }
+
+ switch (fieldSchema.getType()) {
+ case BOOLEAN:
+ return rowData.getBoolean(pos);
+ case INT:
+ return rowData.getInt(pos);
+ case LONG:
+ return rowData.getLong(pos);
+ case FLOAT:
+ return rowData.getFloat(pos);
+ case DOUBLE:
+ return rowData.getDouble(pos);
+ case STRING:
+ return rowData.getString(pos);
+ case BYTES:
+ return rowData.getBinary(pos);
+ case FIXED:
+ return rowData.getBinary(pos);
+ case RECORD:
+ return rowData.getRow(pos, fieldSchema.getFields().size());
+ case ARRAY:
+ return rowData.getArray(pos);
+ case MAP:
+ return rowData.getMap(pos);
+ default:
+ // For complex types, just return as-is
+ return ((GenericRowData) rowData).getField(pos);
Review Comment:
What if the complex type contains a nano timestamp?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2611410642
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/RowDataToAvroGenericRecordConverter.java:
##########
@@ -41,30 +48,116 @@
public class RowDataToAvroGenericRecordConverter implements Function<RowData, GenericRecord> {
private final RowDataToAvroConverters.RowDataToAvroConverter converter;
private final Schema avroSchema;
+ private final Set<Integer> timestampNanosFieldIndices;
+ private final RowData.FieldGetter[] fieldGetters;
- private RowDataToAvroGenericRecordConverter(RowType rowType, Schema avroSchema) {
- this.converter = RowDataToAvroConverters.createConverter(rowType);
+ private RowDataToAvroGenericRecordConverter(
+ RowType converterRowType, Schema avroSchema, Set<Integer> timestampNanosFieldIndices) {
+ this.converter = RowDataToAvroConverters.createConverter(converterRowType);
this.avroSchema = avroSchema;
+ this.timestampNanosFieldIndices = timestampNanosFieldIndices;
+
+ // Pre-create field getters only if there are timestamp-nanos fields
+ // (otherwise we take the early return path and never use them)
+ if (!timestampNanosFieldIndices.isEmpty()) {
+ this.fieldGetters = new RowData.FieldGetter[converterRowType.getFieldCount()];
+ for (int i = 0; i < converterRowType.getFieldCount(); i++) {
+ LogicalType fieldType = converterRowType.getTypeAt(i);
+ this.fieldGetters[i] = RowData.createFieldGetter(fieldType, i);
+ }
+ } else {
+ this.fieldGetters = null;
+ }
}
@Override
public GenericRecord apply(RowData rowData) {
- return (GenericRecord) converter.convert(avroSchema, rowData);
+ // Pre-process: Flink's RowDataToAvroConverters expects Long (nanoseconds) for timestamp-nanos,
+ // but our RowData has TimestampData. Convert TimestampData to Long nanoseconds.
+ if (timestampNanosFieldIndices.isEmpty()) {
+ return (GenericRecord) converter.convert(avroSchema, rowData);
+ }
+
+ // Create a new GenericRowData with Long values for timestamp-nanos fields
+ GenericRowData processedRowData = new GenericRowData(rowData.getArity());
+ processedRowData.setRowKind(rowData.getRowKind());
+
+ for (int i = 0; i < rowData.getArity(); i++) {
+ if (timestampNanosFieldIndices.contains(i)) {
+ // Convert TimestampData to Long nanoseconds
+ if (!rowData.isNullAt(i)) {
+ TimestampData timestampData = rowData.getTimestamp(i, 9);
+ long nanos =
+ timestampData.getMillisecond() * 1_000_000L + timestampData.getNanoOfMillisecond();
+ processedRowData.setField(i, nanos);
+ } else {
+ processedRowData.setField(i, null);
+ }
+ } else {
+ // Copy other fields as-is using pre-created field getter
+ processedRowData.setField(i, fieldGetters[i].getFieldOrNull(rowData));
+ }
+ }
+
+ return (GenericRecord) converter.convert(avroSchema, processedRowData);
}
/** Create a converter based on Iceberg schema */
public static RowDataToAvroGenericRecordConverter fromIcebergSchema(
String tableName, org.apache.iceberg.Schema icebergSchema) {
- RowType rowType = FlinkSchemaUtil.convert(icebergSchema);
+ RowType originalRowType = FlinkSchemaUtil.convert(icebergSchema);
Schema avroSchema = AvroSchemaUtil.convert(icebergSchema, tableName);
- return new RowDataToAvroGenericRecordConverter(rowType, avroSchema);
+
+ // Detect timestamp-nanos fields and create converter RowType with BIGINT for those fields
+ List<LogicalType> converterFields = Lists.newArrayList();
+ Set<Integer> timestampNanosFieldIndices = new java.util.HashSet<>();
+ List<Schema.Field> avroFields = avroSchema.getFields();
+
+ for (int i = 0; i < avroFields.size(); i++) {
+ if (avroFields.get(i).schema().getLogicalType() instanceof LogicalTypes.TimestampNanos) {
+ // Use BIGINT for Flink's converter (it expects Long for timestamp-nanos)
+ converterFields.add(new BigIntType());
+ timestampNanosFieldIndices.add(i);
+ } else {
+ converterFields.add(originalRowType.getTypeAt(i));
+ }
+ }
+
+ RowType converterRowType = RowType.of(converterFields.toArray(new LogicalType[0]));
+ return new RowDataToAvroGenericRecordConverter(
+ converterRowType, avroSchema, timestampNanosFieldIndices);
}
/** Create a mapper based on Avro schema */
public static RowDataToAvroGenericRecordConverter fromAvroSchema(Schema avroSchema) {
DataType dataType = AvroSchemaConverter.convertToDataType(avroSchema.toString());
LogicalType logicalType = TypeConversions.fromDataToLogicalType(dataType);
- RowType rowType = RowType.of(logicalType.getChildren().toArray(new LogicalType[0]));
- return new RowDataToAvroGenericRecordConverter(rowType, avroSchema);
+
+ // Fix up timestamp-nanos fields: Flink's RowDataToAvroConverters expects Long (nanoseconds)
+ // for timestamp-nanos logical type, but our RowData has TimESTAMP(9) with TimestampData.
Review Comment:
nit: `TimESTAMP`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2405272714
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkParquetWriter.java:
##########
@@ -118,4 +121,98 @@ protected void writeAndValidate(Schema schema, List<Record> expectedData) throws
writeAndValidate(binaryRowList, schema);
}
+
+ /** Test that nanosecond precision timestamps are preserved when writing to Parquet files. */
+ @Test
+ public void testNanosecondTimestampPrecision() throws IOException {
Review Comment:
Is the nano timestamp supported by any other file format, like Avro, or ORC?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg IntegrationFlink nanosecond support [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#issuecomment-3364966590
@mxm and @pvary Could you review this pr when you have time ?
I assumed there was nanosecond support for Iceberg V3 with Flink. It turned out we dont. I gave that support.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2439479975
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkValueWriters.java:
##########
@@ -139,9 +139,9 @@ private static class TimestampNanosWriter implements ValueWriter<TimestampData>
@Override
public void write(TimestampData timestampData, Encoder encoder) throws IOException {
- long nanos =
- timestampData.getMillisecond() * 1_000_000 + timestampData.getNanoOfMillisecond();
- encoder.writeLong(nanos);
+ long value =
+ timestampData.getMillisecond() * 1_000_000L + timestampData.getNanoOfMillisecond();
+ encoder.writeLong(value);
Review Comment:
Unneccessary change again
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkValueWriters.java:
##########
@@ -139,9 +139,9 @@ private static class TimestampNanosWriter implements ValueWriter<TimestampData>
@Override
public void write(TimestampData timestampData, Encoder encoder) throws IOException {
- long nanos =
- timestampData.getMillisecond() * 1_000_000 + timestampData.getNanoOfMillisecond();
- encoder.writeLong(nanos);
+ long value =
+ timestampData.getMillisecond() * 1_000_000L + timestampData.getNanoOfMillisecond();
+ encoder.writeLong(value);
Review Comment:
Unnecessary change again
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414560555
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractorEndToEnd.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.reader;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopTableExtension;
+import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.flink.source.FlinkInputFormat;
+import org.apache.iceberg.flink.source.FlinkSource;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * End-to-end tests for ColumnStatsWatermarkExtractor with nanosecond precision timestamps. This
+ * test validates that watermark extraction works correctly with actual data files containing
+ * nanosecond precision timestamps.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestColumnStatsWatermarkExtractorEndToEnd {
+
+ @Parameter(index = 0)
+ private FileFormat format;
+
+ @Parameters(name = "format = {0}")
+ public static Object[][] parameters() {
+ return new Object[][] {{FileFormat.PARQUET}, {FileFormat.AVRO}};
+ }
+
+ private static final String DATABASE = "test_db";
+ private static final String TABLE = "test_watermark_table";
+
+ // Schema with nanosecond precision timestamps for watermark testing
+ // Contains both nanosecond (ns) and microsecond (micros) precision timestamps
+ private static final Schema NANOSECOND_WATERMARK_SCHEMA =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "event_time_ns", Types.TimestampNanoType.withoutZone()),
+ required(3, "event_time_ns_tz", Types.TimestampNanoType.withZone()),
+ required(4, "event_time_micros", Types.TimestampType.withoutZone()),
+ required(5, "data", Types.StringType.get()));
+
+ @TempDir protected Path temporaryDirectory;
+
+ @RegisterExtension
+ private static final HadoopTableExtension TABLE_EXTENSION =
+ HadoopTableExtension.withFormatVersion3(DATABASE, TABLE, NANOSECOND_WATERMARK_SCHEMA);
+
+ /**
+ * Tests that ColumnStatsWatermarkExtractor can be instantiated with both nanosecond and
+ * microsecond precision timestamp columns. This verifies that the extractor correctly recognizes
+ * and accepts both TIMESTAMP_NANO and TIMESTAMP column types.
+ */
+ @Test
+ public void testWatermarkExtractorCreationWithNanosecondTimestamps() {
+ Table table = TABLE_EXTENSION.table();
+
+ // Test that we can create watermark extractors for nanosecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorNs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+ assertThat(extractorNs).isNotNull();
+
+ ColumnStatsWatermarkExtractor extractorNsTz =
+ new ColumnStatsWatermarkExtractor(
+ table.schema(), "event_time_ns_tz", TimeUnit.MICROSECONDS);
+ assertThat(extractorNsTz).isNotNull();
+
+ // Test that we can still create extractors for microsecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorMicros =
+ new ColumnStatsWatermarkExtractor(
+ table.schema(), "event_time_micros", TimeUnit.MICROSECONDS);
+ assertThat(extractorMicros).isNotNull();
Review Comment:
This could not be null.
Are you trying to test if the previous line did not throw an error? Then use `assertDoesNotThrow`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414420129
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/AvroGenericRecordToRowDataMapper.java:
##########
@@ -41,13 +45,114 @@
public class AvroGenericRecordToRowDataMapper implements MapFunction<GenericRecord, RowData> {
private final AvroToRowDataConverters.AvroToRowDataConverter converter;
+ private final Schema avroSchema;
+ private final Function<GenericRecord, RowData> customConverter;
- AvroGenericRecordToRowDataMapper(RowType rowType) {
+ AvroGenericRecordToRowDataMapper(RowType rowType, Schema avroSchema) {
+ this.avroSchema = avroSchema;
this.converter = AvroToRowDataConverters.createRowConverter(rowType);
+ // Check if we need custom conversion for timestamp-nanos
+ this.customConverter = needsCustomConversion(avroSchema) ? this::convertWithNanos : null;
+ }
+
+ private boolean needsCustomConversion(Schema schema) {
+ for (Schema.Field field : schema.getFields()) {
+ if (field.schema().getLogicalType() instanceof LogicalTypes.TimestampNanos) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private RowData convertWithNanos(GenericRecord genericRecord) {
+ RowData baseRowData = (RowData) converter.convert(genericRecord);
+ GenericRowData result = new GenericRowData(baseRowData.getArity());
Review Comment:
I don't really like this solution. We creating 2 `RowData` objects for a single conversion.
Can we do this with a single conversion?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414547146
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java:
##########
@@ -137,12 +137,30 @@ public Type visit(TimeType timeType) {
@Override
public Type visit(TimestampType timestampType) {
- return Types.TimestampType.withoutZone();
+ int precision = timestampType.getPrecision();
+ if (precision > 9) {
+ throw new IllegalArgumentException(
+ "Unsupported timestamp precision: " + precision + ". Maximum supported precision is 9.");
+ }
+ if (precision == 9) {
+ return Types.TimestampNanoType.withoutZone();
+ } else {
Review Comment:
Are precision 7, 8 are valid values?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2409142786
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractorEndToEnd.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.reader;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopTableExtension;
+import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.flink.source.FlinkInputFormat;
+import org.apache.iceberg.flink.source.FlinkSource;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * End-to-end tests for ColumnStatsWatermarkExtractor with nanosecond precision timestamps. This
+ * test validates that watermark extraction works correctly with actual data files containing
+ * nanosecond precision timestamps.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestColumnStatsWatermarkExtractorEndToEnd {
+
+ @Parameter(index = 0)
+ private FileFormat format;
+
+ @Parameters(name = "format = {0}")
+ public static Object[][] parameters() {
+ return new Object[][] {{FileFormat.PARQUET}, {FileFormat.AVRO}};
+ }
+
+ private static final String DATABASE = "test_db";
+ private static final String TABLE = "test_watermark_table";
+
+ // Schema with nanosecond precision timestamps for watermark testing
+ private static final Schema NANOSECOND_WATERMARK_SCHEMA =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "event_time_ns", Types.TimestampNanoType.withoutZone()),
+ required(3, "event_time_ns_tz", Types.TimestampNanoType.withZone()),
+ required(4, "event_time_us", Types.TimestampType.withoutZone()),
+ required(5, "data", Types.StringType.get()));
+
+ @TempDir protected Path temporaryDirectory;
+
+ @RegisterExtension
+ private static final HadoopTableExtension TABLE_EXTENSION =
+ HadoopTableExtension.withFormatVersion3(DATABASE, TABLE, NANOSECOND_WATERMARK_SCHEMA);
+
+ @Test
+ public void testWatermarkExtractorCreationWithNanosecondTimestamps() {
+ Table table = TABLE_EXTENSION.table();
+
+ // Test that we can create watermark extractors for nanosecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorNs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+ assertThat(extractorNs).isNotNull();
+
+ ColumnStatsWatermarkExtractor extractorNsTz =
+ new ColumnStatsWatermarkExtractor(
+ table.schema(), "event_time_ns_tz", TimeUnit.MICROSECONDS);
+ assertThat(extractorNsTz).isNotNull();
+
+ // Test that we can still create extractors for microsecond timestamp columns
+ ColumnStatsWatermarkExtractor extractorUs =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_us", TimeUnit.MICROSECONDS);
+ assertThat(extractorUs).isNotNull();
+ }
+
+ @TestTemplate
+ public void testWatermarkExtractionWithRealData() throws Exception {
+ Table table = TABLE_EXTENSION.table();
+
+ // Generate test data with nanosecond precision timestamps
+ List<Record> testRecords = RandomGenericData.generate(NANOSECOND_WATERMARK_SCHEMA, 5, 42L);
+
+ // Write data to the table
+ new GenericAppenderHelper(table, format, temporaryDirectory).appendToTable(testRecords);
+
+ // Create watermark extractor for nanosecond timestamp column
+ ColumnStatsWatermarkExtractor extractor =
+ new ColumnStatsWatermarkExtractor(table.schema(), "event_time_ns", TimeUnit.MICROSECONDS);
+
+ // Create a mock split with the actual table data
+ // Note: In a real scenario, this would be created by the Flink source system
+ // For this test, we'll validate that the extractor can be created and configured correctly
+ assertThat(extractor).isNotNull();
+
+ // Verify that the table has the expected schema
+ Schema tableSchema = table.schema();
+ assertThat(tableSchema.findField("event_time_ns")).isNotNull();
+ assertThat(tableSchema.findField("event_time_ns_tz")).isNotNull();
+ assertThat(tableSchema.findField("event_time_us")).isNotNull();
+
+ // Verify that the timestamp fields are of the correct types
+ Types.NestedField eventTimeNsField = tableSchema.findField("event_time_ns");
+ assertThat(eventTimeNsField.type()).isInstanceOf(Types.TimestampNanoType.class);
+
+ Types.NestedField eventTimeNsTzField = tableSchema.findField("event_time_ns_tz");
+ assertThat(eventTimeNsTzField.type()).isInstanceOf(Types.TimestampNanoType.class);
+
+ Types.NestedField eventTimeUsField = tableSchema.findField("event_time_us");
+ assertThat(eventTimeUsField.type()).isInstanceOf(Types.TimestampType.class);
+ }
+
+ @TestTemplate
+ public void testDataScanningWithNanosecondTimestamps() throws Exception {
+ Table table = TABLE_EXTENSION.table();
+
+ // Generate test data
+ List<Record> expectedRecords = RandomGenericData.generate(NANOSECOND_WATERMARK_SCHEMA, 3, 123L);
+
+ // Write data to the table
+ new GenericAppenderHelper(table, format, temporaryDirectory).appendToTable(expectedRecords);
+
+ // Test that we can scan the data back
+ FlinkSource.Builder builder =
+ FlinkSource.forRowData().table(table).tableLoader(TABLE_EXTENSION.tableLoader());
+
+ List<Row> actualRows = runFormat(builder.buildFormat());
+
+ // Verify that we can read the data back correctly
+ assertThat(actualRows).hasSize(expectedRecords.size());
+
+ // The data should be readable, indicating that nanosecond timestamps are properly handled
+ // in the read path as well
+ TestHelpers.assertRecords(actualRows, expectedRecords, NANOSECOND_WATERMARK_SCHEMA);
+ }
+
+ @Test
+ public void testWatermarkExtractorWithLongColumn() {
Review Comment:
`testDataScanningWithNanosecondTimestamps` tests nanosecond column extraction
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2409133898
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java:
##########
@@ -137,12 +137,20 @@ public Type visit(TimeType timeType) {
@Override
public Type visit(TimestampType timestampType) {
- return Types.TimestampType.withoutZone();
+ if (timestampType.getPrecision() == 9) {
+ return Types.TimestampNanoType.withoutZone();
+ } else {
+ return Types.TimestampType.withoutZone();
+ }
Review Comment:
No This will be behaviour change. Today Flink Iceberg support Timestamp 0 -> 6 and all maps to microsecond.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414422551
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/RowDataToAvroGenericRecordConverter.java:
##########
@@ -41,17 +44,114 @@
public class RowDataToAvroGenericRecordConverter implements Function<RowData, GenericRecord> {
private final RowDataToAvroConverters.RowDataToAvroConverter converter;
private final Schema avroSchema;
+ private final RowType rowType;
private RowDataToAvroGenericRecordConverter(RowType rowType, Schema avroSchema) {
this.converter = RowDataToAvroConverters.createConverter(rowType);
this.avroSchema = avroSchema;
+ this.rowType = rowType;
}
@Override
public GenericRecord apply(RowData rowData) {
+ // Check if schema has timestamp-nanos fields
+ if (needsCustomConversion(avroSchema)) {
+ return convertWithNanos(rowData);
+ }
return (GenericRecord) converter.convert(avroSchema, rowData);
}
+ private boolean needsCustomConversion(Schema schema) {
+ for (Schema.Field field : schema.getFields()) {
+ if (field.schema().getLogicalType() instanceof LogicalTypes.TimestampNanos) {
+ return true;
+ }
+ }
+ return false;
Review Comment:
Use stream API
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414550016
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkParquetWriter.java:
##########
@@ -118,4 +121,98 @@ protected void writeAndValidate(Schema schema, List<Record> expectedData) throws
writeAndValidate(binaryRowList, schema);
}
+
+ /** Test that nanosecond precision timestamps are preserved when writing to Parquet files. This */
Review Comment:
fix the comment pls
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2606173101
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/StructRowData.java:
##########
@@ -186,8 +186,31 @@ private BigDecimal getDecimalInternal(int pos) {
@Override
public TimestampData getTimestamp(int pos, int precision) {
- long timeLong = getLong(pos);
- return TimestampData.fromEpochMillis(timeLong / 1000, (int) (timeLong % 1000) * 1000);
+ if (precision == 9) {
+ // Nanosecond precision: get nanoseconds directly from struct
+ Object longVal = struct.get(pos, Object.class);
+ long nanos;
+ if (longVal instanceof Long) {
+ nanos = (long) longVal;
+ } else if (longVal instanceof OffsetDateTime) {
+ nanos = Duration.between(Instant.EPOCH, (OffsetDateTime) longVal).toNanos();
+ } else if (longVal instanceof LocalDateTime) {
+ nanos =
+ Duration.between(Instant.EPOCH, ((LocalDateTime) longVal).atOffset(ZoneOffset.UTC))
+ .toNanos();
+ } else {
+ throw new IllegalStateException(
+ "Unknown type for nanosecond timestamp field. Type name: "
+ + longVal.getClass().getName());
+ }
Review Comment:
nit: newline
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2606170369
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/StructRowData.java:
##########
@@ -186,8 +186,31 @@ private BigDecimal getDecimalInternal(int pos) {
@Override
public TimestampData getTimestamp(int pos, int precision) {
- long timeLong = getLong(pos);
- return TimestampData.fromEpochMillis(timeLong / 1000, (int) (timeLong % 1000) * 1000);
+ if (precision == 9) {
+ // Nanosecond precision: get nanoseconds directly from struct
+ Object longVal = struct.get(pos, Object.class);
+ long nanos;
+ if (longVal instanceof Long) {
+ nanos = (long) longVal;
+ } else if (longVal instanceof OffsetDateTime) {
+ nanos = Duration.between(Instant.EPOCH, (OffsetDateTime) longVal).toNanos();
+ } else if (longVal instanceof LocalDateTime) {
+ nanos =
+ Duration.between(Instant.EPOCH, ((LocalDateTime) longVal).atOffset(ZoneOffset.UTC))
+ .toNanos();
+ } else {
Review Comment:
Is it intentional that we don't handle `LocalDate` and `LocalTime`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2606229679
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/AvroGenericRecordToRowDataMapper.java:
##########
@@ -18,44 +18,119 @@
*/
package org.apache.iceberg.flink.sink;
+import java.util.List;
+import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.formats.avro.AvroToRowDataConverters;
import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
+import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.utils.TypeConversions;
-import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
/**
* This util class converts Avro GenericRecord to Flink RowData. <br>
* <br>
* Internally it uses Flink {@link AvroToRowDataConverters}. Because of the precision difference
* between how Iceberg schema (micro) and Flink {@link AvroToRowDataConverters} (milli) deal with
- * time type, we can't directly use the Avro Schema converted from Iceberg schema via {@link
- * AvroSchemaUtil#convert(org.apache.iceberg.Schema, String)}.
+ * time type, we can't directly use the Avro Schema converted from Iceberg schema. Additionally,
+ * Flink's converter doesn't handle timestamp-nanos logical type properly, so we manually fix those
+ * fields after conversion.
*/
public class AvroGenericRecordToRowDataMapper implements MapFunction<GenericRecord, RowData> {
private final AvroToRowDataConverters.AvroToRowDataConverter converter;
+ private final List<Integer> timestampNanosFieldIndices;
+ private final RowType rowType;
- AvroGenericRecordToRowDataMapper(RowType rowType) {
+ AvroGenericRecordToRowDataMapper(RowType rowType, List<Integer> timestampNanosFieldIndices) {
this.converter = AvroToRowDataConverters.createRowConverter(rowType);
+ this.timestampNanosFieldIndices = timestampNanosFieldIndices;
+ this.rowType = rowType;
}
@Override
public RowData map(GenericRecord genericRecord) throws Exception {
- return (RowData) converter.convert(genericRecord);
+ RowData rowData = (RowData) converter.convert(genericRecord);
+
+ // Post-process: Flink's AvroToRowDataConverters doesn't properly handle timestamp-nanos,
+ // so we need to manually fix the timestamp fields after conversion
+ if (timestampNanosFieldIndices.isEmpty()) {
+ return rowData;
+ }
+
+ // Create a new GenericRowData with corrected timestamp-nanos fields
+ GenericRowData correctedRowData = new GenericRowData(rowData.getArity());
+ correctedRowData.setRowKind(rowData.getRowKind());
+
+ // Copy all fields from the converted RowData
+ for (int i = 0; i < rowData.getArity(); i++) {
+ if (timestampNanosFieldIndices.contains(i)) {
+ // Manually convert timestamp-nanos field from original Avro record
+ Object avroValue = genericRecord.get(i);
+ if (avroValue instanceof Long) {
+ long nanos = (Long) avroValue;
+ long millis = Math.floorDiv(nanos, 1_000_000);
+ int nanosOfMillis = Math.floorMod(nanos, 1_000_000);
+ TimestampData timestampData = TimestampData.fromEpochMillis(millis, nanosOfMillis);
+ correctedRowData.setField(i, timestampData);
+ } else {
+ // If not a Long, copy as-is
+ correctedRowData.setField(i, getFieldValue(rowData, i));
+ }
+ } else {
+ // Copy other fields as-is
+ correctedRowData.setField(i, getFieldValue(rowData, i));
+ }
+ }
+
+ return correctedRowData;
+ }
+
+ private Object getFieldValue(RowData rowData, int pos) {
+ if (rowData.isNullAt(pos)) {
+ return null;
+ }
+ // Use FieldGetter to extract the field value
+ LogicalType fieldType = rowType.getTypeAt(pos);
+ RowData.FieldGetter fieldGetter = RowData.createFieldGetter(fieldType, pos);
Review Comment:
Would it be possible to create the fieldgetters in the constructor when there is a nano type present?
It would save quite a bit of time if we don't recreate them every time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2606231559
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/AvroGenericRecordToRowDataMapper.java:
##########
@@ -18,44 +18,119 @@
*/
package org.apache.iceberg.flink.sink;
+import java.util.List;
+import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.formats.avro.AvroToRowDataConverters;
import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
+import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.utils.TypeConversions;
-import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
/**
* This util class converts Avro GenericRecord to Flink RowData. <br>
* <br>
* Internally it uses Flink {@link AvroToRowDataConverters}. Because of the precision difference
* between how Iceberg schema (micro) and Flink {@link AvroToRowDataConverters} (milli) deal with
- * time type, we can't directly use the Avro Schema converted from Iceberg schema via {@link
- * AvroSchemaUtil#convert(org.apache.iceberg.Schema, String)}.
+ * time type, we can't directly use the Avro Schema converted from Iceberg schema. Additionally,
+ * Flink's converter doesn't handle timestamp-nanos logical type properly, so we manually fix those
+ * fields after conversion.
*/
public class AvroGenericRecordToRowDataMapper implements MapFunction<GenericRecord, RowData> {
private final AvroToRowDataConverters.AvroToRowDataConverter converter;
+ private final List<Integer> timestampNanosFieldIndices;
+ private final RowType rowType;
- AvroGenericRecordToRowDataMapper(RowType rowType) {
+ AvroGenericRecordToRowDataMapper(RowType rowType, List<Integer> timestampNanosFieldIndices) {
this.converter = AvroToRowDataConverters.createRowConverter(rowType);
+ this.timestampNanosFieldIndices = timestampNanosFieldIndices;
+ this.rowType = rowType;
}
@Override
public RowData map(GenericRecord genericRecord) throws Exception {
- return (RowData) converter.convert(genericRecord);
+ RowData rowData = (RowData) converter.convert(genericRecord);
+
+ // Post-process: Flink's AvroToRowDataConverters doesn't properly handle timestamp-nanos,
+ // so we need to manually fix the timestamp fields after conversion
+ if (timestampNanosFieldIndices.isEmpty()) {
+ return rowData;
+ }
+
+ // Create a new GenericRowData with corrected timestamp-nanos fields
+ GenericRowData correctedRowData = new GenericRowData(rowData.getArity());
+ correctedRowData.setRowKind(rowData.getRowKind());
+
+ // Copy all fields from the converted RowData
+ for (int i = 0; i < rowData.getArity(); i++) {
+ if (timestampNanosFieldIndices.contains(i)) {
+ // Manually convert timestamp-nanos field from original Avro record
+ Object avroValue = genericRecord.get(i);
+ if (avroValue instanceof Long) {
+ long nanos = (Long) avroValue;
+ long millis = Math.floorDiv(nanos, 1_000_000);
+ int nanosOfMillis = Math.floorMod(nanos, 1_000_000);
+ TimestampData timestampData = TimestampData.fromEpochMillis(millis, nanosOfMillis);
+ correctedRowData.setField(i, timestampData);
+ } else {
+ // If not a Long, copy as-is
+ correctedRowData.setField(i, getFieldValue(rowData, i));
+ }
+ } else {
+ // Copy other fields as-is
+ correctedRowData.setField(i, getFieldValue(rowData, i));
+ }
+ }
+
+ return correctedRowData;
+ }
+
+ private Object getFieldValue(RowData rowData, int pos) {
+ if (rowData.isNullAt(pos)) {
+ return null;
+ }
+ // Use FieldGetter to extract the field value
+ LogicalType fieldType = rowType.getTypeAt(pos);
+ RowData.FieldGetter fieldGetter = RowData.createFieldGetter(fieldType, pos);
Review Comment:
Also, we can use the `fieldGetter.getFieldOrNull(rowData)` to return null values as well. We don't need to create our separate if for this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2405261294
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java:
##########
@@ -459,8 +460,9 @@ private TimestampNanoDataWriter(ColumnDescriptor desc) {
@Override
public void write(int repetitionLevel, TimestampData value) {
- column.writeLong(
- repetitionLevel, value.getMillisecond() * 1_000_000L + value.getNanoOfMillisecond());
+ // For nanosecond precision timestamps, convert to nanoseconds
+ long nanos = value.getMillisecond() * 1_000_000L + value.getNanoOfMillisecond();
+ column.writeLong(repetitionLevel, nanos);
Review Comment:
Could we use the same pattern than the original write for the millisecond timestamp?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2409136987
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestColumnStatsWatermarkExtractorEndToEnd.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source.reader;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.data.GenericAppenderHelper;
+import org.apache.iceberg.data.RandomGenericData;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.HadoopTableExtension;
+import org.apache.iceberg.flink.TestHelpers;
+import org.apache.iceberg.flink.source.FlinkInputFormat;
+import org.apache.iceberg.flink.source.FlinkSource;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * End-to-end tests for ColumnStatsWatermarkExtractor with nanosecond precision timestamps. This
+ * test validates that watermark extraction works correctly with actual data files containing
+ * nanosecond precision timestamps.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestColumnStatsWatermarkExtractorEndToEnd {
+
+ @Parameter(index = 0)
+ private FileFormat format;
+
+ @Parameters(name = "format = {0}")
+ public static Object[][] parameters() {
+ return new Object[][] {{FileFormat.PARQUET}, {FileFormat.AVRO}};
+ }
+
+ private static final String DATABASE = "test_db";
+ private static final String TABLE = "test_watermark_table";
+
+ // Schema with nanosecond precision timestamps for watermark testing
+ private static final Schema NANOSECOND_WATERMARK_SCHEMA =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "event_time_ns", Types.TimestampNanoType.withoutZone()),
+ required(3, "event_time_ns_tz", Types.TimestampNanoType.withZone()),
+ required(4, "event_time_us", Types.TimestampType.withoutZone()),
Review Comment:
this means µs microsecond symbol. I changed with actual name :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add Nanosecond Precision Support for Flink-Iceberg Integration [iceberg]
Posted by "talatuyarer (via GitHub)" <gi...@apache.org>.
talatuyarer commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2409135447
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkParquetWriter.java:
##########
@@ -118,4 +121,98 @@ protected void writeAndValidate(Schema schema, List<Record> expectedData) throws
writeAndValidate(binaryRowList, schema);
}
+
+ /** Test that nanosecond precision timestamps are preserved when writing to Parquet files. */
+ @Test
+ public void testNanosecondTimestampPrecision() throws IOException {
Review Comment:
AVRO supports but not ORC. At least i did not look at it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org