> O createStreamOperator(
@Override
public Class getStreamOperatorClass(ClassLoader classLoader) {
- return StreamingReaderOperator.class;
+ return UnkeyedStreamingReaderOperator.class;
}
}
diff --git a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/UnkeyedStreamingReaderOperator.java b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/UnkeyedStreamingReaderOperator.java
new file mode 100644
index 0000000000..7d4310e358
--- /dev/null
+++ b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/UnkeyedStreamingReaderOperator.java
@@ -0,0 +1,175 @@
+/*
+ * 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.table;
+
+import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
+import org.apache.amoro.shade.guava32.com.google.common.collect.Lists;
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.runtime.state.JavaSerializer;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamSourceContexts;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.util.function.ThrowingRunnable;
+import org.apache.iceberg.flink.source.FlinkInputFormat;
+import org.apache.iceberg.flink.source.FlinkInputSplit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Queue;
+
+/**
+ * Reader operator for Amoro's unkeyed Iceberg file source.
+ *
+ * This mirrors Iceberg's non-public {@code StreamingReaderOperator} behavior while allowing
+ * Amoro to create the {@code FlinkInputFormat} through {@link
+ * org.apache.amoro.flink.interceptor.ProxyFactory} at operator runtime. This keeps the
+ * Kerberos-aware file IO wrapper without reflectively depending on Iceberg's package-private
+ * operator factory or private constructor.
+ */
+public class UnkeyedStreamingReaderOperator extends AbstractStreamOperator
+ implements OneInputStreamOperator {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UnkeyedStreamingReaderOperator.class);
+
+ private final MailboxExecutor executor;
+ private FlinkInputFormat format;
+ private transient SourceFunction.SourceContext sourceContext;
+ private transient ListState inputSplitsState;
+ private transient Queue splits;
+ private transient SplitState currentSplitState;
+
+ public UnkeyedStreamingReaderOperator(
+ FlinkInputFormat format, ProcessingTimeService timeService, MailboxExecutor mailboxExecutor) {
+ this.format = Preconditions.checkNotNull(format, "The InputFormat should not be null.");
+ this.processingTimeService = timeService;
+ this.executor =
+ Preconditions.checkNotNull(mailboxExecutor, "The mailboxExecutor should not be null.");
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ inputSplitsState =
+ context
+ .getOperatorStateStore()
+ .getListState(new ListStateDescriptor<>("splits", new JavaSerializer<>()));
+ currentSplitState = SplitState.IDLE;
+ splits = Lists.newLinkedList();
+ if (context.isRestored()) {
+ int taskIdx = getRuntimeContext().getIndexOfThisSubtask();
+ LOG.info("Restoring state for the {} (taskIdx: {}).", getClass().getSimpleName(), taskIdx);
+ for (FlinkInputSplit split : inputSplitsState.get()) {
+ splits.add(split);
+ }
+ }
+
+ sourceContext =
+ StreamSourceContexts.getSourceContext(
+ getOperatorConfig().getTimeCharacteristic(),
+ getProcessingTimeService(),
+ new Object(),
+ output,
+ getRuntimeContext().getExecutionConfig().getAutoWatermarkInterval(),
+ -1L,
+ true);
+ enqueueProcessSplits();
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ inputSplitsState.clear();
+ inputSplitsState.addAll(Lists.newArrayList(splits));
+ }
+
+ @Override
+ public void processElement(StreamRecord element) {
+ splits.add(element.getValue());
+ enqueueProcessSplits();
+ }
+
+ private void enqueueProcessSplits() {
+ if (currentSplitState == SplitState.IDLE && !splits.isEmpty()) {
+ currentSplitState = SplitState.RUNNING;
+ executor.execute(
+ (ThrowingRunnable) this::processSplits, getClass().getSimpleName());
+ }
+ }
+
+ private void processSplits() throws IOException {
+ FlinkInputSplit split = splits.poll();
+ if (split == null) {
+ currentSplitState = SplitState.IDLE;
+ return;
+ }
+
+ format.open(split);
+ try {
+ RowData next = null;
+ while (!format.reachedEnd()) {
+ next = format.nextRecord(next);
+ sourceContext.collect(next);
+ }
+ } finally {
+ currentSplitState = SplitState.IDLE;
+ format.close();
+ }
+ enqueueProcessSplits();
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) {}
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ if (format != null) {
+ format.close();
+ format.closeInputFormat();
+ format = null;
+ }
+ sourceContext = null;
+ }
+
+ @Override
+ public void finish() throws Exception {
+ super.finish();
+ output.close();
+ if (sourceContext != null) {
+ sourceContext.emitWatermark(Watermark.MAX_WATERMARK);
+ sourceContext.close();
+ sourceContext = null;
+ }
+ }
+
+ private enum SplitState {
+ IDLE,
+ RUNNING
+ }
+}
diff --git a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/util/FlinkClassReflectionUtil.java b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/util/FlinkClassReflectionUtil.java
deleted file mode 100644
index 6b181b510d..0000000000
--- a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/util/FlinkClassReflectionUtil.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.util;
-
-import org.apache.flink.api.connector.source.ReaderOutput;
-import org.apache.flink.streaming.api.operators.source.ProgressiveTimestampsAndWatermarks;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.annotation.Nullable;
-
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-
-/** A util class to handle the reflection operation of Flink class. */
-public class FlinkClassReflectionUtil {
-
- public static final Logger LOG = LoggerFactory.getLogger(FlinkClassReflectionUtil.class);
-
- public static Object getSplitLocalOutput(ReaderOutput readerOutput) {
- if (readerOutput == null) {
- return null;
- }
- try {
- return ReflectionUtil.getField(
- (Class) ProgressiveTimestampsAndWatermarks.class.getDeclaredClasses()[2],
- readerOutput,
- "splitLocalOutputs");
- } catch (Exception e) {
- LOG.warn("extract internal watermark error", e);
- }
- return null;
- }
-
- public static void emitPeriodWatermark(@Nullable Object splitLocalOutput) {
- if (splitLocalOutput == null) {
- return;
- }
- try {
- Method method =
- ProgressiveTimestampsAndWatermarks.class.getDeclaredClasses()[1].getDeclaredMethod(
- "emitPeriodicWatermark");
- method.setAccessible(true);
- method.invoke(splitLocalOutput);
- } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
- LOG.warn("no method found", e);
- }
- }
-}
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 adfd4383b9..76a4555e6b 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
@@ -20,31 +20,31 @@
import org.apache.amoro.flink.interceptor.ProxyFactory;
import org.apache.amoro.io.AuthenticatedFileIO;
-import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
-import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.logical.RowType;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.flink.FlinkConfigOptions;
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.hadoop.Util;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.util.ThreadPools;
import java.lang.reflect.Constructor;
-import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.util.HashMap;
import java.util.List;
@@ -52,8 +52,6 @@
/** An util generates Apache Iceberg writer and committer operator w */
public class IcebergClassUtil {
- private static final String ICEBERG_SCAN_CONTEXT_CLASS =
- "org.apache.iceberg.flink.source.ScanContext";
private static final String ICEBERG_PARTITION_SELECTOR_CLASS =
"org.apache.iceberg.flink.sink.PartitionKeySelector";
private static final String ICEBERG_FILE_COMMITTER_CLASS =
@@ -66,7 +64,6 @@ public static KeySelector newPartitionKeySelector(
try {
Class clazz = forName(ICEBERG_PARTITION_SELECTOR_CLASS);
Constructor c = clazz.getConstructor(PartitionSpec.class, Schema.class, RowType.class);
- c.setAccessible(true);
return (KeySelector) c.newInstance(spec, schema, flinkSchema);
} catch (NoSuchMethodException
| IllegalAccessException
@@ -129,51 +126,22 @@ public static ProxyFactory getIcebergStreamWriterProxyFa
new Object[] {fullTableName, taskWriterFactory});
}
- public static StreamingReaderOperator newStreamingReaderOperator(
- FlinkInputFormat format, ProcessingTimeService timeService, MailboxExecutor mailboxExecutor) {
- try {
- Constructor c =
- StreamingReaderOperator.class.getDeclaredConstructor(
- FlinkInputFormat.class, ProcessingTimeService.class, MailboxExecutor.class);
- c.setAccessible(true);
- return c.newInstance(format, timeService, mailboxExecutor);
- } catch (IllegalAccessException
- | NoSuchMethodException
- | InvocationTargetException
- | InstantiationException e) {
- throw new RuntimeException(e);
- }
- }
-
- public static FlinkInputFormat getInputFormat(OneInputStreamOperatorFactory operatorFactory) {
- try {
- Class[] classes = StreamingReaderOperator.class.getDeclaredClasses();
- Class clazz = null;
- for (Class c : classes) {
- if ("OperatorFactory".equals(c.getSimpleName())) {
- clazz = c;
- break;
- }
- }
- Field field = clazz.getDeclaredField("format");
- field.setAccessible(true);
- return (FlinkInputFormat) (field.get(operatorFactory));
- } catch (IllegalAccessException | NoSuchFieldException e) {
- throw new RuntimeException(e);
- }
- }
-
public static ProxyFactory getInputFormatProxyFactory(
- OneInputStreamOperatorFactory operatorFactory,
+ TableLoader tableLoader,
+ Table table,
AuthenticatedFileIO authenticatedFileIO,
- Schema tableSchema) {
- FlinkInputFormat inputFormat = getInputFormat(operatorFactory);
- TableLoader tableLoader =
- ReflectionUtil.getField(FlinkInputFormat.class, inputFormat, "tableLoader");
- FileIO io = ReflectionUtil.getField(FlinkInputFormat.class, inputFormat, "io");
- EncryptionManager encryption =
- ReflectionUtil.getField(FlinkInputFormat.class, inputFormat, "encryption");
- Object context = ReflectionUtil.getField(FlinkInputFormat.class, inputFormat, "context");
+ Schema tableSchema,
+ Schema projectedSchema,
+ ReadableConfig flinkConf,
+ Map properties,
+ List filters,
+ long limit,
+ Long startSnapshotId) {
+ FileIO io = table.io();
+ EncryptionManager encryption = table.encryption();
+ ScanContext context =
+ buildScanContext(
+ table, projectedSchema, flinkConf, properties, filters, limit, startSnapshotId);
return ProxyUtil.getProxyFactory(
FlinkInputFormat.class,
@@ -184,6 +152,39 @@ public static ProxyFactory getInputFormatProxyFactory(
new Object[] {tableLoader, tableSchema, io, encryption, context});
}
+ private static ScanContext buildScanContext(
+ Table table,
+ Schema projectedSchema,
+ ReadableConfig flinkConf,
+ Map properties,
+ List filters,
+ long limit,
+ Long startSnapshotId) {
+ ScanContext.Builder contextBuilder =
+ ScanContext.builder().resolveConfig(table, properties, flinkConf);
+ contextBuilder.exposeLocality(isLocalityEnabled(table, flinkConf));
+ if (projectedSchema != null) {
+ contextBuilder.project(projectedSchema);
+ }
+ if (filters != null) {
+ contextBuilder.filters(filters);
+ }
+ contextBuilder.limit(limit);
+ if (startSnapshotId != null) {
+ contextBuilder.startSnapshotId(startSnapshotId);
+ }
+ return contextBuilder.build();
+ }
+
+ private static boolean isLocalityEnabled(Table table, ReadableConfig flinkConf) {
+ Boolean localityEnabled =
+ flinkConf.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_EXPOSE_SPLIT_LOCALITY_INFO);
+ if (localityEnabled != null && !localityEnabled) {
+ return false;
+ }
+ return Util.mayHaveBlockLocations(table.io(), table.location());
+ }
+
private static Class forName(String className) {
try {
return Class.forName(className);
@@ -193,22 +194,10 @@ private static Class forName(String className) {
}
public static SourceFunction getSourceFunction(AbstractUdfStreamOperator source) {
- try {
- Field field = AbstractUdfStreamOperator.class.getDeclaredField("userFunction");
- field.setAccessible(true);
- return (SourceFunction) (field.get(source));
- } catch (IllegalAccessException | NoSuchFieldException e) {
- throw new RuntimeException(e);
- }
+ return (SourceFunction) source.getUserFunction();
}
public static void clean(StreamExecutionEnvironment env) {
- try {
- Field field = StreamExecutionEnvironment.class.getDeclaredField("transformations");
- field.setAccessible(true);
- ((List) (field.get(env))).clear();
- } catch (IllegalAccessException | NoSuchFieldException e) {
- throw new RuntimeException(e);
- }
+ env.getTransformations().clear();
}
}
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 192e95801c..31aca4cd2b 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
@@ -43,7 +43,6 @@
import org.apache.amoro.flink.util.CompatibleFlinkPropertyUtil;
import org.apache.amoro.flink.util.IcebergClassUtil;
import org.apache.amoro.flink.util.MixedFormatUtils;
-import org.apache.amoro.flink.util.ProxyUtil;
import org.apache.amoro.table.DistributionHashMode;
import org.apache.amoro.table.MixedTable;
import org.apache.amoro.table.TableProperties;
@@ -435,10 +434,7 @@ public static OneInputStreamOperator createFileCommitter
return null;
}
tableLoader.switchLoadInternalTableForKeyedTable(MixedFormatUtils.isToBase(overwrite));
- return (OneInputStreamOperator)
- ProxyUtil.getProxy(
- IcebergClassUtil.newIcebergFilesCommitter(
- tableLoader, overwrite, branch, spec, mixedTable.io()),
- mixedTable.io());
+ return IcebergClassUtil.newIcebergFilesCommitter(
+ tableLoader, overwrite, branch, spec, mixedTable.io());
}
}
diff --git a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/TestWatermark.java b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/TestWatermark.java
index ae0dbe8c77..35b789bb87 100644
--- a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/TestWatermark.java
+++ b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/table/TestWatermark.java
@@ -30,16 +30,8 @@
import org.apache.amoro.flink.util.TestUtil;
import org.apache.amoro.table.KeyedTable;
import org.apache.amoro.table.TableIdentifier;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.testutils.CommonTestUtils;
-import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
-import org.apache.flink.streaming.api.operators.ChainingStrategy;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.api.DataTypes;
-import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.data.GenericRowData;
@@ -57,8 +49,6 @@
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.time.LocalDateTime;
import java.util.ArrayList;
@@ -68,12 +58,8 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
public class TestWatermark extends FlinkTestBase {
- public static final Logger LOG = LoggerFactory.getLogger(TestWatermark.class);
-
@Rule public TemporaryFolder tempFolder = new TemporaryFolder();
private static final String DB = TableTestHelper.TEST_TABLE_ID.getDatabase();
@@ -147,17 +133,17 @@ public void testWatermark() throws Exception {
"create table d (tt as cast(op_time as timestamp(3)), watermark for tt as tt) like %s",
table);
- Table source = getTableEnv().sqlQuery("select is_true from d");
-
- WatermarkTestOperator op = new WatermarkTestOperator();
- getTableEnv()
- .toRetractStream(source, RowData.class)
- .transform("test watermark", TypeInformation.of(RowData.class), op);
- getEnv().executeAsync("test watermark");
-
- op.waitWatermark();
-
- Assert.assertTrue(op.watermark > Long.MIN_VALUE);
+ // This query verifies that a table with watermark definition can still be consumed
+ // correctly. We intentionally avoid waiting on an async watermark callback here because
+ // that path depends on internal source/operator timing and can hang the test without
+ // revealing a user-visible regression.
+ TableResult result = exec("select is_true from d");
+ CommonTestUtils.waitUntilJobManagerIsInitialized(
+ () -> result.getJobClient().get().getJobStatus().get());
+ try (CloseableIterator iterator = result.collect()) {
+ Assert.assertEquals(Row.of(true), iterator.next());
+ }
+ result.getJobClient().ifPresent(TestUtil::cancelJob);
}
@Test
@@ -226,34 +212,4 @@ public void testSelectWatermarkField() throws Exception {
expected.add(new Object[] {true, LocalDateTime.parse("2022-06-17T10:08:11")});
Assert.assertEquals(DataUtil.toRowSet(expected), actual);
}
-
- public static class WatermarkTestOperator extends AbstractStreamOperator
- implements OneInputStreamOperator, RowData> {
-
- private static final long serialVersionUID = 1L;
- public long watermark;
- private static final CompletableFuture waitWatermark = new CompletableFuture<>();
-
- public WatermarkTestOperator() {
- super();
- chainingStrategy = ChainingStrategy.ALWAYS;
- }
-
- private void waitWatermark() throws InterruptedException, ExecutionException {
- waitWatermark.get();
- }
-
- @Override
- public void processElement(StreamRecord> element) throws Exception {
- output.collect(element.asRecord());
- }
-
- @Override
- public void processWatermark(Watermark mark) throws Exception {
- LOG.info("processWatermark: {}", mark);
- watermark = mark.getTimestamp();
- waitWatermark.complete(null);
- super.processWatermark(mark);
- }
- }
}
diff --git a/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/util/TestIcebergClassUtil.java b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/util/TestIcebergClassUtil.java
new file mode 100644
index 0000000000..0e654f2a07
--- /dev/null
+++ b/amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/test/java/org/apache/amoro/flink/util/TestIcebergClassUtil.java
@@ -0,0 +1,75 @@
+/*
+ * 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.util;
+
+import org.apache.amoro.io.AuthenticatedFileIO;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.flink.sink.FlinkWriteResult;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class TestIcebergClassUtil {
+
+ @Test
+ public void testIcebergFilesCommitterRunsThroughAuthenticatedFileIO() {
+ CountingAuthenticatedFileIO fileIO = new CountingAuthenticatedFileIO();
+ OneInputStreamOperator committer =
+ IcebergClassUtil.newIcebergFilesCommitter(
+ null, false, null, PartitionSpec.unpartitioned(), fileIO);
+
+ committer.toString();
+
+ Assert.assertEquals(1, fileIO.doAsCalls.get());
+ }
+
+ private static class CountingAuthenticatedFileIO implements AuthenticatedFileIO {
+ private final AtomicInteger doAsCalls = new AtomicInteger();
+
+ @Override
+ public T doAs(Callable callable) {
+ doAsCalls.incrementAndGet();
+ try {
+ return callable.call();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public InputFile newInputFile(String path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public OutputFile newOutputFile(String path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void deleteFile(String path) {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
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 b71cf8496f..eb5722a51c 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
@@ -80,6 +80,7 @@
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
@@ -172,26 +173,27 @@ public void testHasCaughtUp() throws Exception {
LocalDateTime.parse("2022-06-18 10:10:11", dtf)
});
List catchUpExpects = new LinkedList<>();
+ LocalDateTime catchUpBaseTime = LocalDateTime.now().truncatedTo(ChronoUnit.MICROS);
catchUpExpects.add(
new Object[] {
1000014,
"d",
- LocalDateTime.now().minusSeconds(3).toEpochSecond(ZoneOffset.UTC),
- LocalDateTime.now().minusSeconds(3)
+ catchUpBaseTime.minusSeconds(3).toEpochSecond(ZoneOffset.UTC),
+ catchUpBaseTime.minusSeconds(3)
});
catchUpExpects.add(
new Object[] {
1000021,
"d",
- LocalDateTime.now().minusSeconds(2).toEpochSecond(ZoneOffset.UTC),
- LocalDateTime.now().minusSeconds(2)
+ catchUpBaseTime.minusSeconds(2).toEpochSecond(ZoneOffset.UTC),
+ catchUpBaseTime.minusSeconds(2)
});
catchUpExpects.add(
new Object[] {
1000015,
"e",
- LocalDateTime.now().minusSeconds(1).toEpochSecond(ZoneOffset.UTC),
- LocalDateTime.now().minusSeconds(1)
+ catchUpBaseTime.minusSeconds(1).toEpochSecond(ZoneOffset.UTC),
+ catchUpBaseTime.minusSeconds(1)
});
expects.addAll(catchUpExpects);
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 7cf7a99eb5..d7f5430299 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
@@ -100,9 +100,12 @@ public TestMixedFormatFileWriter(boolean isKeyed, boolean submitEmptySnapshots)
boolean submitEmptySnapshots,
Long restoredCheckpointId)
throws Exception {
- tableLoader.open();
- MixedTable mixedTable = tableLoader.loadMixedFormatTable();
- mixedTable.properties().put(SUBMIT_EMPTY_SNAPSHOTS.key(), String.valueOf(submitEmptySnapshots));
+ HashMap extraProperties = new HashMap<>();
+ extraProperties.put(SUBMIT_EMPTY_SNAPSHOTS.key(), String.valueOf(submitEmptySnapshots));
+ MixedFormatTableLoader writerTableLoader =
+ tableLoader.copyWithFlinkTableProperties(extraProperties);
+ writerTableLoader.open();
+ MixedTable mixedTable = writerTableLoader.loadMixedFormatTable();
MixedFormatFileWriter streamWriter =
FlinkSink.createFileWriter(
@@ -110,7 +113,7 @@ public TestMixedFormatFileWriter(boolean isKeyed, boolean submitEmptySnapshots)
null,
false,
(RowType) FLINK_SCHEMA.toRowDataType().getLogicalType(),
- tableLoader);
+ writerTableLoader);
TestOneInputStreamOperatorIntern harness =
new TestOneInputStreamOperatorIntern<>(
streamWriter, 1, 1, 0, restoredCheckpointId, new TestGlobalAggregateManager());
diff --git a/amoro-format-mixed/amoro-mixed-trino/pom.xml b/amoro-format-mixed/amoro-mixed-trino/pom.xml
index 202d74a0b6..a91f845fe1 100644
--- a/amoro-format-mixed/amoro-mixed-trino/pom.xml
+++ b/amoro-format-mixed/amoro-mixed-trino/pom.xml
@@ -653,7 +653,6 @@
17
- sun
diff --git a/amoro-format-paimon/pom.xml b/amoro-format-paimon/pom.xml
index 87b29c4282..ff98e9559f 100644
--- a/amoro-format-paimon/pom.xml
+++ b/amoro-format-paimon/pom.xml
@@ -78,4 +78,13 @@