This is an automated email from the ASF dual-hosted git repository.
czy006 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 759168052 [AMORO-4162] Upgrade Iceberg dependency from 1.6.1 to 1.7.2
(#4163)
759168052 is described below
commit 75916805291773dad5caa1c47a32dc2295e80351
Author: Jiwon Park <[email protected]>
AuthorDate: Tue Apr 7 22:19:18 2026 +0900
[AMORO-4162] Upgrade Iceberg dependency from 1.6.1 to 1.7.2 (#4163)
* Upgrade Iceberg from 1.6.1 to 1.7.2
Bump iceberg.version in root pom.xml and update dependency manifests
and documentation references.
Notable transitive dependency changes:
- Avro 1.11.3 → 1.12.0
- httpclient5 5.3.1 → 5.4, httpcore5 5.2.4 → 5.3
- New: failsafe 3.3.2
Signed-off-by: Jiwon Park <[email protected]>
* Adapt to Iceberg 1.7.x API changes
- Migrate WriteResult → FlinkWriteResult in Flink sink pipeline
- Add FieldGetter[] parameter to RowDataUtil.clone() in RowDataRecordFactory
- Migrate PruneColumns from ParquetTypeVisitor to TypeWithSchemaVisitor
- Fix INT96 timestamp min/max statistics byte-order inversion
- Fork MiniClusterResource (removed in Iceberg 1.7.0) into test package
- Update MiniClusterResource imports in all test classes
Signed-off-by: Jiwon Park <[email protected]>
* Add tests for Iceberg 1.7.2 API migration
- TestAdaptHiveParquetSchemaUtil: verify pruneColumns with
TypeWithSchemaVisitor correctly prunes columns using expected schema
- TestAdaptHiveParquetUtil: verify INT96 timestamp min/max statistics
are correctly re-ordered when byte-wise ordering disagrees with
chronological ordering
Signed-off-by: Jiwon Park <[email protected]>
---------
Signed-off-by: Jiwon Park <[email protected]>
Co-authored-by: Xu Bai <[email protected]>
---
.../read/hybrid/reader/RowDataRecordFactory.java | 13 +-
.../apache/amoro/flink/util/IcebergClassUtil.java | 10 +-
.../org/apache/amoro/flink/write/FlinkSink.java | 10 +-
.../amoro/flink/write/MixedFormatFileWriter.java | 10 +-
.../java/org/apache/amoro/flink/FlinkTestBase.java | 1 -
.../apache/amoro/flink/MiniClusterResource.java | 45 ++++
.../amoro/flink/catalog/TestMixedCatalog.java | 2 +-
.../amoro/flink/table/AmoroCatalogITCaseBase.java | 2 +-
.../amoro/flink/table/CatalogITCaseBase.java | 2 +-
.../amoro/flink/table/TestUnkeyedOverwrite.java | 2 +-
.../amoro/flink/write/TestAutomaticLogWriter.java | 16 +-
.../flink/write/TestMixedFormatFileCommitter.java | 28 +--
.../flink/write/TestMixedFormatFileWriter.java | 65 ++++--
.../parquet/AdaptHiveParquetSchemaUtil.java | 4 +-
.../iceberg/parquet/AdaptHiveParquetUtil.java | 28 ++-
.../parquet/TestAdaptHiveParquetSchemaUtil.java | 118 ++++++++++
.../iceberg/parquet/TestAdaptHiveParquetUtil.java | 256 +++++++++++++++++++++
dev/deps/dependencies-hadoop-2-spark-3.3 | 35 +--
dev/deps/dependencies-hadoop-3-spark-3.5 | 35 +--
docs/engines/flink/flink-cdc-ingestion.md | 4 +-
pom.xml | 2 +-
21 files changed, 583 insertions(+), 105 deletions(-)
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/reader/RowDataRecordFactory.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/reader/RowDataRecordFactory.java
index 024c65f80..0793617b0 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/reader/RowDataRecordFactory.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/hybrid/reader/RowDataRecordFactory.java
@@ -29,10 +29,12 @@ import org.apache.iceberg.flink.data.RowDataUtil;
class RowDataRecordFactory implements RecordFactory<RowData> {
private final RowType rowType;
private final TypeSerializer[] fieldSerializers;
+ private final RowData.FieldGetter[] fieldGetters;
RowDataRecordFactory(RowType rowType) {
this.rowType = rowType;
this.fieldSerializers = createFieldSerializers(rowType);
+ this.fieldGetters = createFieldGetters(rowType);
}
static TypeSerializer[] createFieldSerializers(RowType rowType) {
@@ -41,6 +43,14 @@ class RowDataRecordFactory implements RecordFactory<RowData>
{
.toArray(TypeSerializer[]::new);
}
+ static RowData.FieldGetter[] createFieldGetters(RowType rowType) {
+ RowData.FieldGetter[] fieldGetters = new
RowData.FieldGetter[rowType.getFieldCount()];
+ for (int i = 0; i < rowType.getFieldCount(); i++) {
+ fieldGetters[i] = RowData.createFieldGetter(rowType.getTypeAt(i), i);
+ }
+ return fieldGetters;
+ }
+
@Override
public RowData[] createBatch(int batchSize) {
RowData[] arr = new RowData[batchSize];
@@ -57,6 +67,7 @@ class RowDataRecordFactory implements RecordFactory<RowData> {
// Clone method will allocate a new GenericRowData object
// if the target object is NOT a GenericRowData.
// So we should always set the clone return value back to the array.
- batch[position] = RowDataUtil.clone(from, batch[position], rowType,
fieldSerializers);
+ batch[position] =
+ RowDataUtil.clone(from, batch[position], rowType, fieldSerializers,
fieldGetters);
}
}
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/util/IcebergClassUtil.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/util/IcebergClassUtil.java
index 8dcc3eb1b..adfd4383b 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/util/IcebergClassUtil.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/util/IcebergClassUtil.java
@@ -35,12 +35,12 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.encryption.EncryptionManager;
import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.sink.FlinkWriteResult;
import org.apache.iceberg.flink.sink.TaskWriterFactory;
import org.apache.iceberg.flink.source.FlinkInputFormat;
import org.apache.iceberg.flink.source.ScanContext;
import org.apache.iceberg.flink.source.StreamingReaderOperator;
import org.apache.iceberg.io.FileIO;
-import org.apache.iceberg.io.WriteResult;
import org.apache.iceberg.util.ThreadPools;
import java.lang.reflect.Constructor;
@@ -76,7 +76,7 @@ public class IcebergClassUtil {
}
}
- public static OneInputStreamOperator<WriteResult, Void>
newIcebergFilesCommitter(
+ public static OneInputStreamOperator<FlinkWriteResult, Void>
newIcebergFilesCommitter(
TableLoader tableLoader, boolean replacePartitions, String branch,
PartitionSpec spec) {
try {
Class<?> clazz = forName(ICEBERG_FILE_COMMITTER_CLASS);
@@ -89,7 +89,7 @@ public class IcebergClassUtil {
String.class,
PartitionSpec.class);
c.setAccessible(true);
- return (OneInputStreamOperator<WriteResult, Void>)
+ return (OneInputStreamOperator<FlinkWriteResult, Void>)
c.newInstance(
tableLoader,
replacePartitions,
@@ -105,13 +105,13 @@ public class IcebergClassUtil {
}
}
- public static OneInputStreamOperator<WriteResult, Void>
newIcebergFilesCommitter(
+ public static OneInputStreamOperator<FlinkWriteResult, Void>
newIcebergFilesCommitter(
TableLoader tableLoader,
boolean replacePartitions,
String branch,
PartitionSpec spec,
AuthenticatedFileIO authenticatedFileIO) {
- OneInputStreamOperator<WriteResult, Void> obj =
+ OneInputStreamOperator<FlinkWriteResult, Void> obj =
newIcebergFilesCommitter(tableLoader, replacePartitions, branch, spec);
return (OneInputStreamOperator) ProxyUtil.getProxy(obj,
authenticatedFileIO);
}
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/FlinkSink.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/FlinkSink.java
index 38f0ea532..192e95801 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/FlinkSink.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/FlinkSink.java
@@ -65,8 +65,8 @@ import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.sink.FlinkWriteResult;
import org.apache.iceberg.flink.sink.TaskWriterFactory;
-import org.apache.iceberg.io.WriteResult;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.util.PropertyUtil;
import org.slf4j.Logger;
@@ -153,7 +153,7 @@ public class FlinkSink {
DataStream<RowData> input,
MixedFormatLogWriter logWriter,
MixedFormatFileWriter fileWriter,
- OneInputStreamOperator<WriteResult, Void> committer,
+ OneInputStreamOperator<FlinkWriteResult, Void> committer,
int writeOperatorParallelism,
MetricsGenerator metricsGenerator,
String emitMode) {
@@ -161,7 +161,7 @@ public class FlinkSink {
input
.transform(
MixedFormatWriter.class.getName(),
- TypeExtractor.createTypeInfo(WriteResult.class),
+ TypeExtractor.createTypeInfo(FlinkWriteResult.class),
new MixedFormatWriter<>(logWriter, fileWriter,
metricsGenerator))
.name(String.format("MixedFormatWriter %s(%s)", table.name(),
emitMode))
.setParallelism(writeOperatorParallelism);
@@ -414,7 +414,7 @@ public class FlinkSink {
return new MixedFormatRowDataTaskWriterFactory(mixedTable, flinkSchema,
overwrite);
}
- public static OneInputStreamOperator<WriteResult, Void> createFileCommitter(
+ public static OneInputStreamOperator<FlinkWriteResult, Void>
createFileCommitter(
MixedTable mixedTable,
MixedFormatTableLoader tableLoader,
boolean overwrite,
@@ -424,7 +424,7 @@ public class FlinkSink {
mixedTable, tableLoader, overwrite, branch, spec,
MIXED_FORMAT_EMIT_FILE);
}
- public static OneInputStreamOperator<WriteResult, Void> createFileCommitter(
+ public static OneInputStreamOperator<FlinkWriteResult, Void>
createFileCommitter(
MixedTable mixedTable,
MixedFormatTableLoader tableLoader,
boolean overwrite,
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/MixedFormatFileWriter.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/MixedFormatFileWriter.java
index da7bf7285..a7f8e02d3 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/MixedFormatFileWriter.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/MixedFormatFileWriter.java
@@ -34,6 +34,7 @@ import
org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
+import org.apache.iceberg.flink.sink.FlinkWriteResult;
import org.apache.iceberg.flink.sink.TaskWriterFactory;
import org.apache.iceberg.io.TaskWriter;
import org.apache.iceberg.io.WriteResult;
@@ -46,8 +47,8 @@ import java.util.stream.Collectors;
import java.util.stream.IntStream;
/** This is mixed-format table includes writing file data to un keyed table
and keyed table. */
-public class MixedFormatFileWriter extends AbstractStreamOperator<WriteResult>
- implements OneInputStreamOperator<RowData, WriteResult>, BoundedOneInput {
+public class MixedFormatFileWriter extends
AbstractStreamOperator<FlinkWriteResult>
+ implements OneInputStreamOperator<RowData, FlinkWriteResult>,
BoundedOneInput {
private static final long serialVersionUID = 1L;
private static final Logger LOG =
LoggerFactory.getLogger(MixedFormatFileWriter.class);
@@ -61,6 +62,7 @@ public class MixedFormatFileWriter extends
AbstractStreamOperator<WriteResult>
private transient TaskWriter<RowData> writer;
private transient int subTaskId;
+ private transient long currentCheckpointId;
private transient int attemptId;
/**
* Load table in runtime, because that table's refresh method will be
invoked in serialization.
@@ -140,6 +142,7 @@ public class MixedFormatFileWriter extends
AbstractStreamOperator<WriteResult>
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ this.currentCheckpointId = checkpointId;
table
.io()
.doAs(
@@ -153,6 +156,7 @@ public class MixedFormatFileWriter extends
AbstractStreamOperator<WriteResult>
@Override
public void endInput() throws Exception {
+ this.currentCheckpointId = Long.MAX_VALUE;
table
.io()
.doAs(
@@ -205,7 +209,7 @@ public class MixedFormatFileWriter extends
AbstractStreamOperator<WriteResult>
if (shouldEmit(writeResult)) {
// Only emit a non-empty WriteResult to committer operator, thus
avoiding submitting too much
// empty snapshots.
- output.collect(new StreamRecord<>(writeResult));
+ output.collect(new StreamRecord<>(new
FlinkWriteResult(currentCheckpointId, writeResult)));
}
}
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/FlinkTestBase.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/FlinkTestBase.java
index cd3501a07..dbf8238e7 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/FlinkTestBase.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/FlinkTestBase.java
@@ -64,7 +64,6 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.IdentityPartitionConverters;
import org.apache.iceberg.data.Record;
-import org.apache.iceberg.flink.MiniClusterResource;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.TaskWriter;
import org.apache.iceberg.io.WriteResult;
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/MiniClusterResource.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/MiniClusterResource.java
new file mode 100644
index 000000000..e3fab2eaf
--- /dev/null
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/MiniClusterResource.java
@@ -0,0 +1,45 @@
+/*
+ * 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.amoro.flink;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+
+/** Forked from org.apache.iceberg.flink.MiniClusterResource which was removed
in Iceberg 1.7.0. */
+public class MiniClusterResource {
+
+ private static final int DEFAULT_TM_NUM = 1;
+ private static final int DEFAULT_PARALLELISM = 4;
+
+ public static final Configuration DISABLE_CLASSLOADER_CHECK_CONFIG =
+ new Configuration().set(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
+
+ private MiniClusterResource() {}
+
+ public static MiniClusterWithClientResource
createWithClassloaderCheckDisabled() {
+ return new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(DEFAULT_TM_NUM)
+ .setNumberSlotsPerTaskManager(DEFAULT_PARALLELISM)
+ .setConfiguration(DISABLE_CLASSLOADER_CHECK_CONFIG)
+ .build());
+ }
+}
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/catalog/TestMixedCatalog.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/catalog/TestMixedCatalog.java
index b0a009251..3536511d8 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/catalog/TestMixedCatalog.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/catalog/TestMixedCatalog.java
@@ -33,6 +33,7 @@ import org.apache.amoro.TableFormat;
import org.apache.amoro.TableTestHelper;
import org.apache.amoro.catalog.BasicCatalogTestHelper;
import org.apache.amoro.catalog.CatalogTestBase;
+import org.apache.amoro.flink.MiniClusterResource;
import org.apache.amoro.flink.catalog.factories.CatalogFactoryOptions;
import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
@@ -51,7 +52,6 @@ import
org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
import org.apache.flink.util.CollectionUtil;
-import org.apache.iceberg.flink.MiniClusterResource;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/AmoroCatalogITCaseBase.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/AmoroCatalogITCaseBase.java
index 7b5d72363..6ab283e09 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/AmoroCatalogITCaseBase.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/AmoroCatalogITCaseBase.java
@@ -21,6 +21,7 @@ package org.apache.amoro.flink.table;
import static
org.apache.flink.table.api.config.TableConfigOptions.TABLE_DYNAMIC_TABLE_OPTIONS_ENABLED;
import org.apache.amoro.TestAms;
+import org.apache.amoro.flink.MiniClusterResource;
import org.apache.amoro.formats.AmoroCatalogTestBase;
import org.apache.amoro.formats.AmoroCatalogTestHelper;
import org.apache.amoro.hive.TestHMS;
@@ -36,7 +37,6 @@ import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.iceberg.flink.MiniClusterResource;
import org.junit.ClassRule;
import java.io.IOException;
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/CatalogITCaseBase.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/CatalogITCaseBase.java
index 2e10a280c..84500df82 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/CatalogITCaseBase.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/CatalogITCaseBase.java
@@ -23,6 +23,7 @@ import static
org.apache.flink.table.api.config.TableConfigOptions.TABLE_DYNAMIC
import org.apache.amoro.TableTestHelper;
import org.apache.amoro.catalog.CatalogTestHelper;
import org.apache.amoro.catalog.TableTestBase;
+import org.apache.amoro.flink.MiniClusterResource;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.StateBackend;
@@ -35,7 +36,6 @@ import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.iceberg.flink.MiniClusterResource;
import org.junit.ClassRule;
public abstract class CatalogITCaseBase extends TableTestBase {
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/TestUnkeyedOverwrite.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/TestUnkeyedOverwrite.java
index fcb092e3d..026b20e65 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/TestUnkeyedOverwrite.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/TestUnkeyedOverwrite.java
@@ -24,6 +24,7 @@ import org.apache.amoro.TableTestHelper;
import org.apache.amoro.catalog.BasicCatalogTestHelper;
import org.apache.amoro.catalog.CatalogTestHelper;
import org.apache.amoro.flink.FlinkTestBase;
+import org.apache.amoro.flink.MiniClusterResource;
import org.apache.amoro.flink.util.DataUtil;
import org.apache.amoro.hive.TestHMS;
import org.apache.amoro.hive.catalog.HiveCatalogTestHelper;
@@ -32,7 +33,6 @@ import org.apache.flink.table.api.ApiExpression;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.test.util.MiniClusterWithClientResource;
-import org.apache.iceberg.flink.MiniClusterResource;
import org.junit.After;
import org.junit.Assert;
import org.junit.ClassRule;
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestAutomaticLogWriter.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestAutomaticLogWriter.java
index 37150e356..b71cf8496 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestAutomaticLogWriter.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestAutomaticLogWriter.java
@@ -60,7 +60,7 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.UpdateProperties;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.flink.FlinkSchemaUtil;
-import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.flink.sink.FlinkWriteResult;
import org.apache.iceberg.types.TypeUtil;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.producer.ProducerConfig;
@@ -256,12 +256,12 @@ public class TestAutomaticLogWriter extends FlinkTestBase
{
}
List<Object[]> expects = new LinkedList<>();
- List<WriteResult> results;
+ List<FlinkWriteResult> results;
testKeyedTable.refresh();
Assert.assertFalse(
Boolean.parseBoolean(
testKeyedTable.properties().getOrDefault(LOG_STORE_CATCH_UP.key(),
"false")));
- try (TestOneInputStreamOperatorIntern<RowData, WriteResult> harness =
+ try (TestOneInputStreamOperatorIntern<RowData, FlinkWriteResult> harness =
createSingleProducer(1, jobId, topic, gap)) {
DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd
HH:mm:ss");
expects.add(
@@ -314,7 +314,7 @@ public class TestAutomaticLogWriter extends FlinkTestBase {
// check expects accuracy.
Assert.assertEquals(3, results.size());
- results.forEach(result -> Assert.assertEquals(1,
result.dataFiles().length));
+ results.forEach(result -> Assert.assertEquals(1,
result.writeResult().dataFiles().length));
List<Object[]> expected = isGapNone ? expects : expects.subList(2,
expects.size());
checkLogstoreDataAccuracy(topic, expected);
testKeyedTable.refresh();
@@ -350,7 +350,7 @@ public class TestAutomaticLogWriter extends FlinkTestBase {
.collect(Collectors.toList()));
}
- public TestOneInputStreamOperatorIntern<RowData, WriteResult>
createSingleProducer(
+ public TestOneInputStreamOperatorIntern<RowData, FlinkWriteResult>
createSingleProducer(
int maxParallelism, byte[] jobId, String topic, Duration
writeLogstoreWatermarkGap)
throws Exception {
return createProducer(
@@ -364,7 +364,7 @@ public class TestAutomaticLogWriter extends FlinkTestBase {
writeLogstoreWatermarkGap);
}
- private TestOneInputStreamOperatorIntern<RowData, WriteResult>
createProducer(
+ private TestOneInputStreamOperatorIntern<RowData, FlinkWriteResult>
createProducer(
int maxParallelism,
int parallelism,
int subTaskId,
@@ -402,10 +402,10 @@ public class TestAutomaticLogWriter extends FlinkTestBase
{
(RowType) FLINK_SCHEMA.toRowDataType().getLogicalType(),
tableLoader);
- MixedFormatWriter<WriteResult> mixedFormatWriter =
+ MixedFormatWriter<FlinkWriteResult> mixedFormatWriter =
new MixedFormatWriter<>(automaticLogWriter, streamWriter,
metricsGenerator);
- TestOneInputStreamOperatorIntern<RowData, WriteResult> harness =
+ TestOneInputStreamOperatorIntern<RowData, FlinkWriteResult> harness =
new TestOneInputStreamOperatorIntern<>(
mixedFormatWriter,
maxParallelism,
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestMixedFormatFileCommitter.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestMixedFormatFileCommitter.java
index 6b3b572e3..bc0bfe823 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestMixedFormatFileCommitter.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestMixedFormatFileCommitter.java
@@ -36,8 +36,8 @@ import org.apache.flink.types.RowKind;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.TableScan;
+import org.apache.iceberg.flink.sink.FlinkWriteResult;
import org.apache.iceberg.io.CloseableIterable;
-import org.apache.iceberg.io.WriteResult;
import org.junit.Assert;
import org.junit.Test;
@@ -52,15 +52,15 @@ public class TestMixedFormatFileCommitter extends
FlinkTestBase {
new BasicTableTestHelper(true, true));
}
- public OneInputStreamOperatorTestHarness<WriteResult, Void>
createMixedFormatFileCommitter(
+ public OneInputStreamOperatorTestHarness<FlinkWriteResult, Void>
createMixedFormatFileCommitter(
MixedFormatTableLoader tableLoader,
MixedTable table,
OperatorSubtaskState operatorSubtaskState)
throws Exception {
- OneInputStreamOperator<WriteResult, Void> committer =
+ OneInputStreamOperator<FlinkWriteResult, Void> committer =
FlinkSink.createFileCommitter(
table, tableLoader, false, SnapshotRef.MAIN_BRANCH, table.spec());
- OneInputStreamOperatorTestHarness<WriteResult, Void> harness =
+ OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> harness =
new OneInputStreamOperatorTestHarness<>(committer, 1, 1, 0);
harness.setup();
@@ -93,19 +93,19 @@ public class TestMixedFormatFileCommitter extends
FlinkTestBase {
tableLoader = MixedFormatTableLoader.of(TableTestHelper.TEST_TABLE_ID,
catalogBuilder);
KeyedTable table =
MixedFormatUtils.loadMixedTable(tableLoader).asKeyedTable();
- List<WriteResult> completedFiles = prepareChangeFiles();
+ List<FlinkWriteResult> completedFiles = prepareChangeFiles();
OperatorSubtaskState snapshot;
long checkpoint = 1;
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> testHarness =
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> testHarness
=
createMixedFormatFileCommitter(tableLoader, table, null)) {
- for (WriteResult completedFile : completedFiles) {
+ for (FlinkWriteResult completedFile : completedFiles) {
testHarness.processElement(new StreamRecord<>(completedFile));
}
snapshot = testHarness.snapshot(checkpoint, System.currentTimeMillis());
}
- try (OneInputStreamOperatorTestHarness<WriteResult, Void> testHarness =
+ try (OneInputStreamOperatorTestHarness<FlinkWriteResult, Void> testHarness
=
createMixedFormatFileCommitter(tableLoader, table, snapshot)) {
testHarness.notifyOfCompletedCheckpoint(checkpoint);
}
@@ -113,10 +113,10 @@ public class TestMixedFormatFileCommitter extends
FlinkTestBase {
checkChangeFiles(7, 9, table);
}
- private List<WriteResult> prepareChangeFiles() throws Exception {
- List<WriteResult> changeFiles;
+ private List<FlinkWriteResult> prepareChangeFiles() throws Exception {
+ List<FlinkWriteResult> changeFiles;
long checkpointId = 1L;
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
TestMixedFormatFileWriter.createMixedFormatStreamWriter(tableLoader)) {
// The first checkpoint
testHarness.processElement(createRowData(1, "hello",
"2020-10-11T10:10:11.0"), 1);
@@ -125,7 +125,8 @@ public class TestMixedFormatFileCommitter extends
FlinkTestBase {
testHarness.prepareSnapshotPreBarrier(checkpointId);
Assert.assertEquals(1, testHarness.extractOutputValues().size());
- Assert.assertEquals(3,
testHarness.extractOutputValues().get(0).dataFiles().length);
+ Assert.assertEquals(
+ 3,
testHarness.extractOutputValues().get(0).writeResult().dataFiles().length);
checkpointId = checkpointId + 1;
@@ -143,7 +144,8 @@ public class TestMixedFormatFileCommitter extends
FlinkTestBase {
testHarness.prepareSnapshotPreBarrier(checkpointId);
// testHarness.extractOutputValues() compute the sum
Assert.assertEquals(2, testHarness.extractOutputValues().size());
- Assert.assertEquals(4,
testHarness.extractOutputValues().get(1).dataFiles().length);
+ Assert.assertEquals(
+ 4,
testHarness.extractOutputValues().get(1).writeResult().dataFiles().length);
changeFiles = testHarness.extractOutputValues();
}
return changeFiles;
diff --git
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestMixedFormatFileWriter.java
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestMixedFormatFileWriter.java
index 889fd74e1..7cf7a99eb 100644
---
a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestMixedFormatFileWriter.java
+++
b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/write/TestMixedFormatFileWriter.java
@@ -36,6 +36,7 @@ import org.apache.flink.types.RowKind;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.SerializableTable;
import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.sink.FlinkWriteResult;
import org.apache.iceberg.flink.sink.RowDataTaskWriterFactory;
import org.apache.iceberg.flink.sink.TaskWriterFactory;
import org.apache.iceberg.io.TaskWriter;
@@ -73,18 +74,18 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
this.submitEmptySnapshots = submitEmptySnapshots;
}
- public static OneInputStreamOperatorTestHarness<RowData, WriteResult>
+ public static OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
createMixedFormatStreamWriter(MixedFormatTableLoader tableLoader) throws
Exception {
return createMixedFormatStreamWriter(tableLoader, true, null);
}
- public static OneInputStreamOperatorTestHarness<RowData, WriteResult>
+ public static OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
createMixedFormatStreamWriter(
MixedFormatTableLoader tableLoader,
boolean submitEmptySnapshots,
Long restoredCheckpointId)
throws Exception {
- OneInputStreamOperatorTestHarness<RowData, WriteResult> harness =
+ OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult> harness =
doCreateMixedFormatStreamWriter(tableLoader, submitEmptySnapshots,
restoredCheckpointId);
harness.setup();
@@ -93,7 +94,7 @@ public class TestMixedFormatFileWriter extends FlinkTestBase {
return harness;
}
- public static OneInputStreamOperatorTestHarness<RowData, WriteResult>
+ public static OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
doCreateMixedFormatStreamWriter(
MixedFormatTableLoader tableLoader,
boolean submitEmptySnapshots,
@@ -110,7 +111,7 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
false,
(RowType) FLINK_SCHEMA.toRowDataType().getLogicalType(),
tableLoader);
- TestOneInputStreamOperatorIntern<RowData, WriteResult> harness =
+ TestOneInputStreamOperatorIntern<RowData, FlinkWriteResult> harness =
new TestOneInputStreamOperatorIntern<>(
streamWriter, 1, 1, 0, restoredCheckpointId, new
TestGlobalAggregateManager());
@@ -137,7 +138,7 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
Assume.assumeTrue(isKeyedTable());
tableLoader = MixedFormatTableLoader.of(TableTestHelper.TEST_TABLE_ID,
catalogBuilder);
long checkpointId = 1L;
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createMixedFormatStreamWriter(tableLoader)) {
MixedFormatFileWriter fileWriter = (MixedFormatFileWriter)
testHarness.getOneInputOperator();
Assert.assertNotNull(fileWriter.getWriter());
@@ -149,7 +150,8 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
testHarness.prepareSnapshotPreBarrier(checkpointId);
Assert.assertNull(fileWriter.getWriter());
Assert.assertEquals(1, testHarness.extractOutputValues().size());
- Assert.assertEquals(3,
testHarness.extractOutputValues().get(0).dataFiles().length);
+ Assert.assertEquals(
+ 3,
testHarness.extractOutputValues().get(0).writeResult().dataFiles().length);
checkpointId = checkpointId + 1;
@@ -161,9 +163,9 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
testHarness.prepareSnapshotPreBarrier(checkpointId);
// testHarness.extractOutputValues() calculates the cumulative value
- List<WriteResult> completedFiles = testHarness.extractOutputValues();
+ List<FlinkWriteResult> completedFiles =
testHarness.extractOutputValues();
Assert.assertEquals(2, completedFiles.size());
- Assert.assertEquals(3, completedFiles.get(1).dataFiles().length);
+ Assert.assertEquals(3,
completedFiles.get(1).writeResult().dataFiles().length);
}
}
@@ -173,7 +175,7 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
long timestamp = 1;
tableLoader = MixedFormatTableLoader.of(TableTestHelper.TEST_TABLE_ID,
catalogBuilder);
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createMixedFormatStreamWriter(tableLoader)) {
testHarness.processElement(createRowData(1, "hello",
"2020-10-11T10:10:11.0"), timestamp++);
testHarness.processElement(createRowData(2, "hello",
"2020-10-12T10:10:11.0"), timestamp);
@@ -181,7 +183,13 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
testHarness.prepareSnapshotPreBarrier(checkpointId++);
long expectedDataFiles = 3;
- WriteResult result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ WriteResult result =
+ WriteResult.builder()
+ .addAll(
+ testHarness.extractOutputValues().stream()
+ .map(FlinkWriteResult::writeResult)
+ .collect(java.util.stream.Collectors.toList()))
+ .build();
Assert.assertEquals(0, result.deleteFiles().length);
Assert.assertEquals(expectedDataFiles, result.dataFiles().length);
@@ -189,7 +197,13 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
for (int i = 0; i < 5; i++) {
testHarness.prepareSnapshotPreBarrier(checkpointId++);
- result =
WriteResult.builder().addAll(testHarness.extractOutputValues()).build();
+ result =
+ WriteResult.builder()
+ .addAll(
+ testHarness.extractOutputValues().stream()
+ .map(FlinkWriteResult::writeResult)
+ .collect(java.util.stream.Collectors.toList()))
+ .build();
Assert.assertEquals(0, result.deleteFiles().length);
Assert.assertEquals(expectedDataFiles, result.dataFiles().length);
}
@@ -201,7 +215,7 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
Assume.assumeFalse(isKeyedTable());
tableLoader = MixedFormatTableLoader.of(TableTestHelper.TEST_TABLE_ID,
catalogBuilder);
long checkpointId = 1L;
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createMixedFormatStreamWriter(tableLoader)) {
// The first checkpoint
testHarness.processElement(createRowData(1, "hello",
"2020-10-11T10:10:11.0"), 1);
@@ -210,7 +224,8 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
testHarness.prepareSnapshotPreBarrier(checkpointId);
Assert.assertEquals(1, testHarness.extractOutputValues().size());
- Assert.assertEquals(3,
testHarness.extractOutputValues().get(0).dataFiles().length);
+ Assert.assertEquals(
+ 3,
testHarness.extractOutputValues().get(0).writeResult().dataFiles().length);
checkpointId = checkpointId + 1;
@@ -221,9 +236,9 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
testHarness.prepareSnapshotPreBarrier(checkpointId);
// testHarness.extractOutputValues() calculates the cumulative value
- List<WriteResult> completedFiles = testHarness.extractOutputValues();
+ List<FlinkWriteResult> completedFiles =
testHarness.extractOutputValues();
Assert.assertEquals(2, completedFiles.size());
- Assert.assertEquals(1, completedFiles.get(1).dataFiles().length);
+ Assert.assertEquals(1,
completedFiles.get(1).writeResult().dataFiles().length);
}
}
@@ -232,7 +247,7 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
Assume.assumeTrue(isKeyedTable());
tableLoader = MixedFormatTableLoader.of(TableTestHelper.TEST_TABLE_ID,
catalogBuilder);
long checkpointId = 1L;
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createMixedFormatStreamWriter(tableLoader)) {
// The first checkpoint
testHarness.processElement(
@@ -246,7 +261,8 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
testHarness.prepareSnapshotPreBarrier(checkpointId);
Assert.assertEquals(1, testHarness.extractOutputValues().size());
- Assert.assertEquals(3,
testHarness.extractOutputValues().get(0).dataFiles().length);
+ Assert.assertEquals(
+ 3,
testHarness.extractOutputValues().get(0).writeResult().dataFiles().length);
checkpointId = checkpointId + 1;
@@ -261,7 +277,8 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
testHarness.prepareSnapshotPreBarrier(checkpointId);
// testHarness.extractOutputValues() calculates the cumulative value
Assert.assertEquals(2, testHarness.extractOutputValues().size());
- Assert.assertEquals(3,
testHarness.extractOutputValues().get(1).dataFiles().length);
+ Assert.assertEquals(
+ 3,
testHarness.extractOutputValues().get(1).writeResult().dataFiles().length);
}
}
@@ -270,7 +287,7 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
Assume.assumeTrue(isKeyedTable());
tableLoader = MixedFormatTableLoader.of(TableTestHelper.TEST_TABLE_ID,
catalogBuilder);
long checkpointId = 1L;
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createMixedFormatStreamWriter(tableLoader)) {
// The first checkpoint
testHarness.processElement(
@@ -284,7 +301,8 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
testHarness.prepareSnapshotPreBarrier(checkpointId);
Assert.assertEquals(1, testHarness.extractOutputValues().size());
- Assert.assertEquals(3,
testHarness.extractOutputValues().get(0).dataFiles().length);
+ Assert.assertEquals(
+ 3,
testHarness.extractOutputValues().get(0).writeResult().dataFiles().length);
checkpointId = checkpointId + 1;
@@ -300,7 +318,8 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
testHarness.prepareSnapshotPreBarrier(checkpointId);
// testHarness.extractOutputValues() calculates the cumulative value
Assert.assertEquals(2, testHarness.extractOutputValues().size());
- Assert.assertEquals(3,
testHarness.extractOutputValues().get(1).dataFiles().length);
+ Assert.assertEquals(
+ 3,
testHarness.extractOutputValues().get(1).writeResult().dataFiles().length);
}
}
@@ -310,7 +329,7 @@ public class TestMixedFormatFileWriter extends
FlinkTestBase {
tableLoader = MixedFormatTableLoader.of(TableTestHelper.TEST_TABLE_ID,
catalogBuilder);
long checkpointId = 1L;
long excepted = submitEmptySnapshots ? 1 : 0;
- try (OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness =
+ try (OneInputStreamOperatorTestHarness<RowData, FlinkWriteResult>
testHarness =
createMixedFormatStreamWriter(tableLoader, submitEmptySnapshots,
null)) {
// The first checkpoint
diff --git
a/amoro-format-mixed/amoro-mixed-hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetSchemaUtil.java
b/amoro-format-mixed/amoro-mixed-hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetSchemaUtil.java
index 9ef8c1cad..a3f378e5f 100644
---
a/amoro-format-mixed/amoro-mixed-hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetSchemaUtil.java
+++
b/amoro-format-mixed/amoro-mixed-hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetSchemaUtil.java
@@ -79,7 +79,9 @@ public class AdaptHiveParquetSchemaUtil {
public static MessageType pruneColumns(MessageType fileSchema, Schema
expectedSchema) {
// column order must match the incoming type, so it doesn't matter that
the ids are unordered
Set<Integer> selectedIds = TypeUtil.getProjectedIds(expectedSchema);
- return (MessageType) ParquetTypeVisitor.visit(fileSchema, new
PruneColumns(selectedIds));
+ return (MessageType)
+ TypeWithSchemaVisitor.visit(
+ expectedSchema.asStruct(), fileSchema, new
PruneColumns(selectedIds));
}
/**
diff --git
a/amoro-format-mixed/amoro-mixed-hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetUtil.java
b/amoro-format-mixed/amoro-mixed-hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetUtil.java
index c5563a865..d8c1aa884 100644
---
a/amoro-format-mixed/amoro-mixed-hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetUtil.java
+++
b/amoro-format-mixed/amoro-mixed-hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetUtil.java
@@ -122,14 +122,29 @@ public class AdaptHiveParquetUtil {
Types.NestedField field = fileSchema.findField(fieldId);
if (field != null && stats.hasNonNullValue() &&
shouldStoreBounds(column, fileSchema)) {
// Change for mixed-hive table ⬇
- // Add metrics for int96 type
- Literal<?> min =
+ // Add metrics for int96 type.
+ // For INT96 timestamps, Parquet's byte-wise min/max comparison
does not
+ // match chronological ordering due to little-endian storage of
+ // nanoseconds-of-day. Convert both values first, then determine
the
+ // actual min and max from the converted (microsecond)
representations.
+ Literal<?> val1 =
AdaptHiveParquetConversions.fromParquetPrimitive(
field.type(), column.getPrimitiveType(),
stats.genericGetMin());
- updateMin(lowerBounds, fieldId, field.type(), min, metricsMode);
- Literal<?> max =
+ Literal<?> val2 =
AdaptHiveParquetConversions.fromParquetPrimitive(
field.type(), column.getPrimitiveType(),
stats.genericGetMax());
+ Literal<?> min;
+ Literal<?> max;
+ @SuppressWarnings("unchecked")
+ java.util.Comparator<Object> cmp =
(java.util.Comparator<Object>) val1.comparator();
+ if (isInt96(column) && cmp.compare(val1.value(), val2.value()) >
0) {
+ min = val2;
+ max = val1;
+ } else {
+ min = val1;
+ max = val2;
+ }
+ updateMin(lowerBounds, fieldId, field.type(), min, metricsMode);
updateMax(upperBounds, fieldId, field.type(), max, metricsMode);
// Change for mixed-hive table ⬆
}
@@ -244,6 +259,11 @@ public class AdaptHiveParquetUtil {
return currentType != null && currentType.isPrimitiveType();
}
+ private static boolean isInt96(ColumnChunkMetaData column) {
+ return column.getPrimitiveType().getPrimitiveTypeName()
+ == org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT96;
+ }
+
private static void increment(Map<Integer, Long> columns, int fieldId, long
amount) {
if (columns != null) {
if (columns.containsKey(fieldId)) {
diff --git
a/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/iceberg/parquet/TestAdaptHiveParquetSchemaUtil.java
b/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/iceberg/parquet/TestAdaptHiveParquetSchemaUtil.java
new file mode 100644
index 000000000..c9d1f2cf7
--- /dev/null
+++
b/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/iceberg/parquet/TestAdaptHiveParquetSchemaUtil.java
@@ -0,0 +1,118 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.types.Types;
+import org.apache.parquet.schema.MessageType;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Tests for {@link AdaptHiveParquetSchemaUtil}, specifically verifying that
{@code pruneColumns}
+ * uses {@link TypeWithSchemaVisitor} (Iceberg 1.7+) to correctly prune
columns using the expected
+ * schema. Prior to this migration the method used {@link ParquetTypeVisitor}
which did not pass the
+ * expected schema, causing incorrect results for schemas whose column order
differed from the file
+ * schema.
+ */
+public class TestAdaptHiveParquetSchemaUtil {
+
+ @Test
+ public void testPruneColumnsSelectsSubset() {
+ // Full file schema with 3 columns
+ Schema fullSchema =
+ new Schema(
+ required(1, "id", Types.IntegerType.get()),
+ optional(2, "name", Types.StringType.get()),
+ optional(3, "ts", Types.TimestampType.withoutZone()));
+
+ MessageType fileSchema = AdaptHiveParquetSchemaUtil.convert(fullSchema,
"test");
+
+ // Expected schema requests only 2 of the 3 columns
+ Schema expectedSchema =
+ new Schema(
+ required(1, "id", Types.IntegerType.get()),
+ optional(3, "ts", Types.TimestampType.withoutZone()));
+
+ MessageType pruned = AdaptHiveParquetSchemaUtil.pruneColumns(fileSchema,
expectedSchema);
+
+ Assert.assertEquals(
+ "Pruned schema should contain only the requested columns", 2,
pruned.getFieldCount());
+ Assert.assertTrue("Pruned schema should contain 'id'",
pruned.containsField("id"));
+ Assert.assertTrue("Pruned schema should contain 'ts'",
pruned.containsField("ts"));
+ Assert.assertFalse("Pruned schema should not contain 'name'",
pruned.containsField("name"));
+ }
+
+ @Test
+ public void testPruneColumnsWithNestedStruct() {
+ // File schema with a nested struct
+ Schema fullSchema =
+ new Schema(
+ required(1, "id", Types.IntegerType.get()),
+ optional(
+ 2,
+ "data",
+ Types.StructType.of(
+ required(3, "x", Types.IntegerType.get()),
+ optional(4, "y", Types.StringType.get()))),
+ optional(5, "extra", Types.StringType.get()));
+
+ MessageType fileSchema = AdaptHiveParquetSchemaUtil.convert(fullSchema,
"test");
+
+ // Request only id and the nested struct — the TypeWithSchemaVisitor
approach
+ // correctly preserves nested fields using the expectedSchema tree.
+ Schema expectedSchema =
+ new Schema(
+ required(1, "id", Types.IntegerType.get()),
+ optional(
+ 2,
+ "data",
+ Types.StructType.of(
+ required(3, "x", Types.IntegerType.get()),
+ optional(4, "y", Types.StringType.get()))));
+
+ MessageType pruned = AdaptHiveParquetSchemaUtil.pruneColumns(fileSchema,
expectedSchema);
+
+ Assert.assertEquals(2, pruned.getFieldCount());
+ Assert.assertTrue(pruned.containsField("id"));
+ Assert.assertTrue(pruned.containsField("data"));
+ Assert.assertFalse(pruned.containsField("extra"));
+ }
+
+ @Test
+ public void testPruneColumnsSingleColumn() {
+ Schema fullSchema =
+ new Schema(
+ required(1, "id", Types.IntegerType.get()),
+ optional(2, "name", Types.StringType.get()),
+ optional(3, "value", Types.DoubleType.get()));
+
+ MessageType fileSchema = AdaptHiveParquetSchemaUtil.convert(fullSchema,
"test");
+
+ Schema expectedSchema = new Schema(optional(2, "name",
Types.StringType.get()));
+
+ MessageType pruned = AdaptHiveParquetSchemaUtil.pruneColumns(fileSchema,
expectedSchema);
+
+ Assert.assertEquals(1, pruned.getFieldCount());
+ Assert.assertTrue(pruned.containsField("name"));
+ }
+}
diff --git
a/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/iceberg/parquet/TestAdaptHiveParquetUtil.java
b/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/iceberg/parquet/TestAdaptHiveParquetUtil.java
new file mode 100644
index 000000000..b1b8b9302
--- /dev/null
+++
b/amoro-format-mixed/amoro-mixed-hive/src/test/java/org/apache/iceberg/parquet/TestAdaptHiveParquetUtil.java
@@ -0,0 +1,256 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import org.apache.iceberg.FieldMetrics;
+import org.apache.iceberg.Metrics;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Types;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnPath;
+import org.apache.parquet.hadoop.metadata.FileMetaData;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+/**
+ * Tests for {@link AdaptHiveParquetUtil#footerMetrics}, verifying that INT96
timestamp statistics
+ * are correctly re-ordered.
+ *
+ * <p>Parquet stores INT96 timestamps as 12 bytes (8-byte LE
nanoseconds-of-day + 4-byte LE Julian
+ * day). Byte-wise comparison compares the nanos first, which does not match
chronological order
+ * when timestamps cross day boundaries. Without the fix,
stats.genericGetMin() could be
+ * chronologically *later* than stats.genericGetMax(), causing inverted
lower/upper bounds in
+ * Iceberg metrics.
+ */
+public class TestAdaptHiveParquetUtil {
+
+ private static final int JULIAN_EPOCH_OFFSET = 2_440_588;
+
+ /**
+ * Encodes a unix-epoch-millisecond timestamp into a 12-byte INT96 Parquet
binary (LE
+ * nanoseconds-of-day + LE Julian day).
+ */
+ private static byte[] toInt96Bytes(long epochMillis) {
+ long epochDay = Math.floorDiv(epochMillis, TimeUnit.DAYS.toMillis(1));
+ long milliOfDay = epochMillis - epochDay * TimeUnit.DAYS.toMillis(1);
+ long nanoOfDay = milliOfDay * 1_000_000L;
+ int julianDay = (int) (epochDay + JULIAN_EPOCH_OFFSET);
+
+ ByteBuffer buf = ByteBuffer.allocate(12).order(ByteOrder.LITTLE_ENDIAN);
+ buf.putLong(nanoOfDay);
+ buf.putInt(julianDay);
+ return buf.array();
+ }
+
+ /**
+ * Verifies that footerMetrics swaps INT96 min/max when byte-wise ordering
disagrees with
+ * chronological ordering.
+ *
+ * <p>We construct two timestamps crossing a day boundary:
+ *
+ * <ul>
+ * <li>earlier: 2020-01-01 23:59:59.999 UTC (large nanos-of-day, end of
day)
+ * <li>later: 2020-01-02 00:00:00.001 UTC (small nanos-of-day, start of
next day)
+ * </ul>
+ *
+ * In INT96 little-endian format, the nanos-of-day occupy the first 8 bytes.
The earlier timestamp
+ * has a much larger nanos value (86399999 ms → ~8.6e13 nanos), so byte-wise
it appears "greater"
+ * than the later timestamp (1 ms → 1e6 nanos). Parquet therefore sets
stats.genericGetMin() =
+ * later, stats.genericGetMax() = earlier. The fix detects this inversion
for INT96 columns and
+ * swaps them.
+ */
+ @Test
+ public void testInt96MinMaxSwappedWhenByteOrderReversed() {
+ // 2020-01-01 23:59:59.999 UTC — large nanos-of-day (end of day)
+ long earlierMillis = 1577923199999L;
+ // 2020-01-02 00:00:00.001 UTC — small nanos-of-day (start of next day)
+ long laterMillis = 1577923200001L;
+
+ byte[] earlierBytes = toInt96Bytes(earlierMillis);
+ byte[] laterBytes = toInt96Bytes(laterMillis);
+
+ // Verify our premise: byte-wise, laterBytes < earlierBytes
+ // (because nanos-of-day for 01:00 < nanos-of-day for 23:00, and nanos
come first in LE)
+ Assert.assertTrue(
+ "Precondition: byte-wise earlier > later for these INT96 values",
+ compareBytewise(earlierBytes, laterBytes) > 0);
+
+ // Build Parquet schema with INT96 column carrying Iceberg field id
+ MessageType parquetSchema =
+ new MessageType(
+ "test",
+ Collections.singletonList(
+
org.apache.parquet.schema.Types.required(PrimitiveType.PrimitiveTypeName.INT96)
+ .id(1)
+ .named("ts")));
+
+ // Build statistics: Parquet's byte-wise min = laterBytes, max =
earlierBytes
+ Statistics<?> stats =
+
Statistics.getBuilderForReading(parquetSchema.getType("ts").asPrimitiveType())
+ .withMin(laterBytes)
+ .withMax(earlierBytes)
+ .withNumNulls(0)
+ .build();
+
+ // Column chunk metadata
+ Set<Encoding> encodings = new HashSet<>();
+ encodings.add(Encoding.PLAIN);
+ ColumnChunkMetaData columnMeta =
+ ColumnChunkMetaData.get(
+ ColumnPath.get("ts"),
+ parquetSchema.getType("ts").asPrimitiveType(),
+
org.apache.parquet.hadoop.metadata.CompressionCodecName.UNCOMPRESSED,
+ null,
+ encodings,
+ stats,
+ 0L,
+ 0L,
+ 1L,
+ 100L,
+ 100L);
+
+ BlockMetaData block = new BlockMetaData();
+ block.addColumn(columnMeta);
+ block.setRowCount(1);
+
+ Schema icebergSchema = new Schema(required(1, "ts",
Types.TimestampType.withoutZone()));
+ Map<String, String> keyValueMetadata = new HashMap<>();
+ keyValueMetadata.put("iceberg.schema", SchemaParser.toJson(icebergSchema));
+ FileMetaData fileMetaData = new FileMetaData(parquetSchema,
keyValueMetadata, "test");
+ ParquetMetadata metadata = new ParquetMetadata(fileMetaData,
Collections.singletonList(block));
+
+ // Call footerMetrics
+ Metrics metrics =
+ AdaptHiveParquetUtil.footerMetrics(
+ metadata, Stream.<FieldMetrics<?>>empty(),
MetricsConfig.getDefault(), icebergSchema);
+
+ // Extract lower and upper bounds for field 1 (ts)
+ ByteBuffer lowerBuf = metrics.lowerBounds().get(1);
+ ByteBuffer upperBuf = metrics.upperBounds().get(1);
+
+ Assert.assertNotNull("Lower bound should exist for ts", lowerBuf);
+ Assert.assertNotNull("Upper bound should exist for ts", upperBuf);
+
+ long lowerMicros =
Conversions.fromByteBuffer(Types.TimestampType.withoutZone(), lowerBuf);
+ long upperMicros =
Conversions.fromByteBuffer(Types.TimestampType.withoutZone(), upperBuf);
+
+ Assert.assertTrue(
+ "Lower bound (earlier timestamp) must be <= upper bound (later
timestamp), "
+ + "but got lower="
+ + lowerMicros
+ + ", upper="
+ + upperMicros,
+ lowerMicros <= upperMicros);
+ }
+
+ /**
+ * Verifies that non-INT96 columns are not affected by the swap logic —
their min/max follow
+ * Parquet's original order.
+ */
+ @Test
+ public void testNonInt96ColumnMinMaxPreserved() {
+ PrimitiveType int32Type =
+
org.apache.parquet.schema.Types.required(PrimitiveType.PrimitiveTypeName.INT32)
+ .id(1)
+ .named("id");
+ MessageType parquetSchema = new MessageType("test",
Collections.singletonList(int32Type));
+
+ // min = 10, max = 42
+ byte[] minBytes =
ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN).putInt(10).array();
+ byte[] maxBytes =
ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN).putInt(42).array();
+
+ Statistics<?> stats =
+ Statistics.getBuilderForReading(int32Type)
+ .withMin(minBytes)
+ .withMax(maxBytes)
+ .withNumNulls(0)
+ .build();
+
+ Set<Encoding> encodings = new HashSet<>();
+ encodings.add(Encoding.PLAIN);
+ ColumnChunkMetaData columnMeta =
+ ColumnChunkMetaData.get(
+ ColumnPath.get("id"),
+ int32Type,
+
org.apache.parquet.hadoop.metadata.CompressionCodecName.UNCOMPRESSED,
+ null,
+ encodings,
+ stats,
+ 0L,
+ 0L,
+ 2L,
+ 100L,
+ 100L);
+
+ BlockMetaData block = new BlockMetaData();
+ block.addColumn(columnMeta);
+ block.setRowCount(2);
+
+ Schema icebergSchema = new Schema(required(1, "id",
Types.IntegerType.get()));
+ Map<String, String> keyValueMetadata = new HashMap<>();
+ keyValueMetadata.put("iceberg.schema", SchemaParser.toJson(icebergSchema));
+ FileMetaData fileMetaData = new FileMetaData(parquetSchema,
keyValueMetadata, "test");
+ ParquetMetadata metadata = new ParquetMetadata(fileMetaData,
Collections.singletonList(block));
+
+ Metrics metrics =
+ AdaptHiveParquetUtil.footerMetrics(
+ metadata, Stream.<FieldMetrics<?>>empty(),
MetricsConfig.getDefault(), icebergSchema);
+
+ ByteBuffer lowerBuf = metrics.lowerBounds().get(1);
+ ByteBuffer upperBuf = metrics.upperBounds().get(1);
+ Assert.assertNotNull(lowerBuf);
+ Assert.assertNotNull(upperBuf);
+
+ int lower = Conversions.fromByteBuffer(Types.IntegerType.get(), lowerBuf);
+ int upper = Conversions.fromByteBuffer(Types.IntegerType.get(), upperBuf);
+ Assert.assertEquals(10, lower);
+ Assert.assertEquals(42, upper);
+ }
+
+ private static int compareBytewise(byte[] a, byte[] b) {
+ for (int i = 0; i < Math.min(a.length, b.length); i++) {
+ int cmp = Byte.compareUnsigned(a[i], b[i]);
+ if (cmp != 0) {
+ return cmp;
+ }
+ }
+ return Integer.compare(a.length, b.length);
+ }
+}
diff --git a/dev/deps/dependencies-hadoop-2-spark-3.3
b/dev/deps/dependencies-hadoop-2-spark-3.3
index d8a476f6e..4cfe8086e 100644
--- a/dev/deps/dependencies-hadoop-2-spark-3.3
+++ b/dev/deps/dependencies-hadoop-2-spark-3.3
@@ -31,7 +31,7 @@ auth/2.24.12//auth-2.24.12.jar
aviator/5.9.0//aviator-5.9.0.jar
avro-ipc/1.11.0//avro-ipc-1.11.0.jar
avro-mapred/1.11.0//avro-mapred-1.11.0.jar
-avro/1.11.3//avro-1.11.3.jar
+avro/1.12.0//avro-1.12.0.jar
aws-core/2.24.12//aws-core-2.24.12.jar
aws-json-protocol/2.24.12//aws-json-protocol-2.24.12.jar
aws-query-protocol/2.24.12//aws-query-protocol-2.24.12.jar
@@ -82,6 +82,7 @@ ehcache/3.3.1//ehcache-3.3.1.jar
endpoints-spi/2.24.12//endpoints-spi-2.24.12.jar
error_prone_annotations/2.18.0//error_prone_annotations-2.18.0.jar
eventstream/1.0.1//eventstream-1.0.1.jar
+failsafe/3.3.2//failsafe-3.3.2.jar
failureaccess/1.0.1//failureaccess-1.0.1.jar
flatbuffers-java/23.5.26//flatbuffers-java-23.5.26.jar
flink-annotations/1.20.3//flink-annotations-1.20.3.jar
@@ -145,23 +146,23 @@ http-auth-spi/2.24.12//http-auth-spi-2.24.12.jar
http-auth/2.24.12//http-auth-2.24.12.jar
http-client-spi/2.24.12//http-client-spi-2.24.12.jar
httpclient/4.5.13//httpclient-4.5.13.jar
-httpclient5/5.3.1//httpclient5-5.3.1.jar
+httpclient5/5.4//httpclient5-5.4.jar
httpcore/4.4.13//httpcore-4.4.13.jar
-httpcore5-h2/5.2.4//httpcore5-h2-5.2.4.jar
-httpcore5/5.2.4//httpcore5-5.2.4.jar
-iceberg-aliyun/1.6.1//iceberg-aliyun-1.6.1.jar
-iceberg-api/1.6.1//iceberg-api-1.6.1.jar
-iceberg-arrow/1.6.1//iceberg-arrow-1.6.1.jar
-iceberg-aws/1.6.1//iceberg-aws-1.6.1.jar
-iceberg-bundled-guava/1.6.1//iceberg-bundled-guava-1.6.1.jar
-iceberg-common/1.6.1//iceberg-common-1.6.1.jar
-iceberg-core/1.6.1//iceberg-core-1.6.1.jar
-iceberg-data/1.6.1//iceberg-data-1.6.1.jar
-iceberg-hive-metastore/1.6.1//iceberg-hive-metastore-1.6.1.jar
-iceberg-orc/1.6.1//iceberg-orc-1.6.1.jar
-iceberg-parquet/1.6.1//iceberg-parquet-1.6.1.jar
-iceberg-spark-3.3_2.12/1.6.1//iceberg-spark-3.3_2.12-1.6.1.jar
-iceberg-spark-extensions-3.3_2.12/1.6.1//iceberg-spark-extensions-3.3_2.12-1.6.1.jar
+httpcore5-h2/5.3//httpcore5-h2-5.3.jar
+httpcore5/5.3//httpcore5-5.3.jar
+iceberg-aliyun/1.7.2//iceberg-aliyun-1.7.2.jar
+iceberg-api/1.7.2//iceberg-api-1.7.2.jar
+iceberg-arrow/1.7.2//iceberg-arrow-1.7.2.jar
+iceberg-aws/1.7.2//iceberg-aws-1.7.2.jar
+iceberg-bundled-guava/1.7.2//iceberg-bundled-guava-1.7.2.jar
+iceberg-common/1.7.2//iceberg-common-1.7.2.jar
+iceberg-core/1.7.2//iceberg-core-1.7.2.jar
+iceberg-data/1.7.2//iceberg-data-1.7.2.jar
+iceberg-hive-metastore/1.7.2//iceberg-hive-metastore-1.7.2.jar
+iceberg-orc/1.7.2//iceberg-orc-1.7.2.jar
+iceberg-parquet/1.7.2//iceberg-parquet-1.7.2.jar
+iceberg-spark-3.3_2.12/1.7.2//iceberg-spark-3.3_2.12-1.7.2.jar
+iceberg-spark-extensions-3.3_2.12/1.7.2//iceberg-spark-extensions-3.3_2.12-1.7.2.jar
icu4j/69.1//icu4j-69.1.jar
identity-spi/2.24.12//identity-spi-2.24.12.jar
ipaddress/5.5.1//ipaddress-5.5.1.jar
diff --git a/dev/deps/dependencies-hadoop-3-spark-3.5
b/dev/deps/dependencies-hadoop-3-spark-3.5
index 94f037281..a97b4a692 100644
--- a/dev/deps/dependencies-hadoop-3-spark-3.5
+++ b/dev/deps/dependencies-hadoop-3-spark-3.5
@@ -26,7 +26,7 @@ auth/2.24.12//auth-2.24.12.jar
aviator/5.9.0//aviator-5.9.0.jar
avro-ipc/1.11.5//avro-ipc-1.11.5.jar
avro-mapred/1.11.5//avro-mapred-1.11.5.jar
-avro/1.11.3//avro-1.11.3.jar
+avro/1.12.0//avro-1.12.0.jar
aws-core/2.24.12//aws-core-2.24.12.jar
aws-json-protocol/2.24.12//aws-json-protocol-2.24.12.jar
aws-query-protocol/2.24.12//aws-query-protocol-2.24.12.jar
@@ -75,6 +75,7 @@ eclipse-collections/11.1.0//eclipse-collections-11.1.0.jar
endpoints-spi/2.24.12//endpoints-spi-2.24.12.jar
error_prone_annotations/2.18.0//error_prone_annotations-2.18.0.jar
eventstream/1.0.1//eventstream-1.0.1.jar
+failsafe/3.3.2//failsafe-3.3.2.jar
failureaccess/1.0.1//failureaccess-1.0.1.jar
flatbuffers-java/23.5.26//flatbuffers-java-23.5.26.jar
flatbuffers/1.2.0-3f79e055//flatbuffers-1.2.0-3f79e055.jar
@@ -128,23 +129,23 @@ http-auth-spi/2.24.12//http-auth-spi-2.24.12.jar
http-auth/2.24.12//http-auth-2.24.12.jar
http-client-spi/2.24.12//http-client-spi-2.24.12.jar
httpclient/4.5.14//httpclient-4.5.14.jar
-httpclient5/5.3.1//httpclient5-5.3.1.jar
+httpclient5/5.4//httpclient5-5.4.jar
httpcore/4.4.16//httpcore-4.4.16.jar
-httpcore5-h2/5.2.4//httpcore5-h2-5.2.4.jar
-httpcore5/5.2.4//httpcore5-5.2.4.jar
-iceberg-aliyun/1.6.1//iceberg-aliyun-1.6.1.jar
-iceberg-api/1.6.1//iceberg-api-1.6.1.jar
-iceberg-arrow/1.6.1//iceberg-arrow-1.6.1.jar
-iceberg-aws/1.6.1//iceberg-aws-1.6.1.jar
-iceberg-bundled-guava/1.6.1//iceberg-bundled-guava-1.6.1.jar
-iceberg-common/1.6.1//iceberg-common-1.6.1.jar
-iceberg-core/1.6.1//iceberg-core-1.6.1.jar
-iceberg-data/1.6.1//iceberg-data-1.6.1.jar
-iceberg-hive-metastore/1.6.1//iceberg-hive-metastore-1.6.1.jar
-iceberg-orc/1.6.1//iceberg-orc-1.6.1.jar
-iceberg-parquet/1.6.1//iceberg-parquet-1.6.1.jar
-iceberg-spark-3.5_2.12/1.6.1//iceberg-spark-3.5_2.12-1.6.1.jar
-iceberg-spark-extensions-3.5_2.12/1.6.1//iceberg-spark-extensions-3.5_2.12-1.6.1.jar
+httpcore5-h2/5.3//httpcore5-h2-5.3.jar
+httpcore5/5.3//httpcore5-5.3.jar
+iceberg-aliyun/1.7.2//iceberg-aliyun-1.7.2.jar
+iceberg-api/1.7.2//iceberg-api-1.7.2.jar
+iceberg-arrow/1.7.2//iceberg-arrow-1.7.2.jar
+iceberg-aws/1.7.2//iceberg-aws-1.7.2.jar
+iceberg-bundled-guava/1.7.2//iceberg-bundled-guava-1.7.2.jar
+iceberg-common/1.7.2//iceberg-common-1.7.2.jar
+iceberg-core/1.7.2//iceberg-core-1.7.2.jar
+iceberg-data/1.7.2//iceberg-data-1.7.2.jar
+iceberg-hive-metastore/1.7.2//iceberg-hive-metastore-1.7.2.jar
+iceberg-orc/1.7.2//iceberg-orc-1.7.2.jar
+iceberg-parquet/1.7.2//iceberg-parquet-1.7.2.jar
+iceberg-spark-3.5_2.12/1.7.2//iceberg-spark-3.5_2.12-1.7.2.jar
+iceberg-spark-extensions-3.5_2.12/1.7.2//iceberg-spark-extensions-3.5_2.12-1.7.2.jar
icu4j/69.1//icu4j-69.1.jar
identity-spi/2.24.12//identity-spi-2.24.12.jar
ipaddress/5.5.1//ipaddress-5.5.1.jar
diff --git a/docs/engines/flink/flink-cdc-ingestion.md
b/docs/engines/flink/flink-cdc-ingestion.md
index 09a67d2c1..163fd5759 100644
--- a/docs/engines/flink/flink-cdc-ingestion.md
+++ b/docs/engines/flink/flink-cdc-ingestion.md
@@ -34,7 +34,7 @@ The following example will show how [MySQL
CDC](https://nightlies.apache.org/fli
**Requirements**
-Please add [Flink SQL Connector MySQL
CDC](https://mvnrepository.com/artifact/org.apache.flink/flink-connector-mysql-cdc)
and
[Iceberg](https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-flink-1.18/1.6.0/iceberg-flink-1.18-1.6.0.jar)
Jars to the lib directory of the Flink engine package.
+Please add [Flink SQL Connector MySQL
CDC](https://mvnrepository.com/artifact/org.apache.flink/flink-connector-mysql-cdc)
and
[Iceberg](https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-flink-1.18/1.7.2/iceberg-flink-1.18-1.7.2.jar)
Jars to the lib directory of the Flink engine package.
```sql
CREATE TABLE products (
@@ -114,7 +114,7 @@ The following example will show how to write CDC data from
multiple MySQL tables
**Requirements**
Please add [Flink Connector MySQL
CDC](https://mvnrepository.com/artifact/org.apache.flink/flink-connector-mysql-cdc/3.1.1)
-and
[Iceberg](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-flink-1.18/1.6.0)
dependencies to your
+and
[Iceberg](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-flink-1.18/1.7.2)
dependencies to your
Maven project's pom.xml file.
```java
diff --git a/pom.xml b/pom.xml
index 32a414d27..1b54c24bc 100644
--- a/pom.xml
+++ b/pom.xml
@@ -102,7 +102,7 @@
<maven-clean-plugin.version>3.3.2</maven-clean-plugin.version>
<maven-checkstyle-plugin.version>3.3.1</maven-checkstyle-plugin.version>
- <iceberg.version>1.6.1</iceberg.version>
+ <iceberg.version>1.7.2</iceberg.version>
<paimon.version>1.2.0</paimon.version>
<hive.version>3.1.3</hive.version>
<hadoop.version>3.4.3</hadoop.version>
