From fc6c5f4285098d18cd7f6e81785f59e68a3b6862 Mon Sep 17 00:00:00 2001 From: Danny Chan Date: Mon, 15 Mar 2021 16:02:05 +0800 Subject: [PATCH] [HUDI-1684] Tweak hudi-flink-bundle module pom and reorganize the pacakges for hudi-flink module (#2669) * Add required dependencies for hudi-flink-bundle module * Some packages reorganization of hudi-flink module --- .../FlinkOptions.java | 6 +- .../hudi/schema/FilebasedSchemaProvider.java | 2 +- .../java/org/apache/hudi/sink/CommitSink.java | 1 - .../InstantGenerateOperator.java | 8 +- .../KeyedWriteProcessFunction.java | 6 +- .../KeyedWriteProcessOperator.java | 3 +- .../StreamWriteFunction.java | 5 +- .../StreamWriteOperator.java | 2 +- .../StreamWriteOperatorCoordinator.java | 5 +- .../StreamWriteOperatorFactory.java | 2 +- .../compact/CompactFunction.java | 2 +- .../compact/CompactionCommitEvent.java | 2 +- .../compact/CompactionCommitSink.java | 2 +- .../compact/CompactionPlanEvent.java | 2 +- .../compact/CompactionPlanOperator.java | 2 +- .../event/BatchWriteSuccessEvent.java | 6 +- .../partitioner/BucketAssignFunction.java | 4 +- .../partitioner/BucketAssigner.java | 2 +- .../partitioner/BucketAssigners.java | 4 +- .../delta/DeltaBucketAssigner.java | 4 +- .../JsonStringToHoodieRecordMapFunction.java | 2 +- .../transform/RowDataToHoodieFunction.java | 4 +- .../StreamReadMonitoringFunction.java | 11 +- .../StreamReadOperator.java | 6 +- .../hudi/streamer/HoodieFlinkStreamer.java | 12 +- .../hudi/streamer/HoodieFlinkStreamerV2.java | 8 +- .../HoodieTableFactory.java | 6 +- .../hudi/{sink => table}/HoodieTableSink.java | 20 +-- .../{source => table}/HoodieTableSource.java | 20 +-- .../format/FilePathUtils.java | 4 +- .../{source => table}/format/FormatUtils.java | 6 +- .../format/cow/AbstractColumnReader.java | 2 +- .../format/cow/CopyOnWriteInputFormat.java | 2 +- .../cow/Int64TimestampColumnReader.java | 2 +- .../cow/ParquetColumnarRowSplitReader.java | 6 +- .../format/cow/ParquetDecimalVector.java | 2 +- .../format/cow/ParquetSplitReaderUtil.java | 2 +- .../format/cow/RunLengthDecoder.java | 2 +- .../format/mor/InstantRange.java | 2 +- .../format/mor/MergeOnReadInputFormat.java | 14 +- .../format/mor/MergeOnReadInputSplit.java | 2 +- .../format/mor/MergeOnReadTableState.java | 2 +- .../hudi/util/RowDataToAvroConverters.java | 9 +- .../org/apache/hudi/util/StreamerUtil.java | 2 +- ....apache.flink.table.factories.TableFactory | 2 +- .../{operator => sink}/StreamWriteITCase.java | 22 +-- .../TestStreamWriteOperatorCoordinator.java} | 8 +- .../TestWriteCopyOnWrite.java | 11 +- .../TestWriteMergeOnRead.java | 4 +- .../TestWriteMergeOnReadWithCompact.java | 3 +- .../partitioner/TestBucketAssigner.java | 4 +- ...stJsonStringToHoodieRecordMapFunction.java | 2 +- .../utils/CompactFunctionWrapper.java | 14 +- .../MockFunctionInitializationContext.java | 2 +- .../utils/MockMapState.java | 2 +- .../utils/MockOperatorStateStore.java | 2 +- .../utils/MockStreamingRuntimeContext.java | 2 +- .../utils/StreamWriteFunctionWrapper.java | 15 +- .../TestStreamReadMonitoringFunction.java | 9 +- .../hudi/source/TestStreamReadOperator.java | 16 +- .../HoodieDataSourceITCase.java | 10 +- .../TestHoodieTableFactory.java | 8 +- .../TestHoodieTableSource.java | 10 +- .../format/TestInputFormat.java | 10 +- .../utils/TestConfigurations.java | 4 +- .../hudi/{operator => }/utils/TestData.java | 5 +- .../java/org/apache/hudi/utils/TestUtils.java | 6 +- .../factory/CollectSinkTableFactory.java | 2 +- .../factory/ContinuousFileSourceFactory.java | 2 +- .../utils/source/ContinuousFileSource.java | 2 +- ....apache.flink.table.factories.TableFactory | 2 +- packaging/hudi-flink-bundle/pom.xml | 163 +++++++++++++++++- 72 files changed, 357 insertions(+), 203 deletions(-) rename hudi-flink/src/main/java/org/apache/hudi/{operator => configuration}/FlinkOptions.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/InstantGenerateOperator.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/KeyedWriteProcessFunction.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/KeyedWriteProcessOperator.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/StreamWriteFunction.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/StreamWriteOperator.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/StreamWriteOperatorCoordinator.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/StreamWriteOperatorFactory.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/compact/CompactFunction.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/compact/CompactionCommitEvent.java (97%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/compact/CompactionCommitSink.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/compact/CompactionPlanEvent.java (97%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/compact/CompactionPlanOperator.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/event/BatchWriteSuccessEvent.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/partitioner/BucketAssignFunction.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/partitioner/BucketAssigner.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/partitioner/BucketAssigners.java (93%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/partitioner/delta/DeltaBucketAssigner.java (97%) rename hudi-flink/src/main/java/org/apache/hudi/{source => sink/transform}/JsonStringToHoodieRecordMapFunction.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => sink}/transform/RowDataToHoodieFunction.java (97%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => source}/StreamReadMonitoringFunction.java (97%) rename hudi-flink/src/main/java/org/apache/hudi/{operator => source}/StreamReadOperator.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{factory => table}/HoodieTableFactory.java (96%) rename hudi-flink/src/main/java/org/apache/hudi/{sink => table}/HoodieTableSink.java (89%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/HoodieTableSource.java (96%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/FilePathUtils.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/FormatUtils.java (96%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/cow/AbstractColumnReader.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/cow/CopyOnWriteInputFormat.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/cow/Int64TimestampColumnReader.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/cow/ParquetColumnarRowSplitReader.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/cow/ParquetDecimalVector.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/cow/ParquetSplitReaderUtil.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/cow/RunLengthDecoder.java (99%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/mor/InstantRange.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/mor/MergeOnReadInputFormat.java (97%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/mor/MergeOnReadInputSplit.java (98%) rename hudi-flink/src/main/java/org/apache/hudi/{source => table}/format/mor/MergeOnReadTableState.java (98%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => sink}/StreamWriteITCase.java (95%) rename hudi-flink/src/test/java/org/apache/hudi/{operator/StreamWriteOperatorCoordinatorTest.java => sink/TestStreamWriteOperatorCoordinator.java} (96%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => sink}/TestWriteCopyOnWrite.java (98%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => sink}/TestWriteMergeOnRead.java (97%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => sink}/TestWriteMergeOnReadWithCompact.java (95%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => sink}/partitioner/TestBucketAssigner.java (98%) rename hudi-flink/src/test/java/org/apache/hudi/{source => sink/transform}/TestJsonStringToHoodieRecordMapFunction.java (98%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => sink}/utils/CompactFunctionWrapper.java (91%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => sink}/utils/MockFunctionInitializationContext.java (97%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => sink}/utils/MockMapState.java (98%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => sink}/utils/MockOperatorStateStore.java (99%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => sink}/utils/MockStreamingRuntimeContext.java (99%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => sink}/utils/StreamWriteFunctionWrapper.java (94%) rename hudi-flink/src/test/java/org/apache/hudi/{source => table}/HoodieDataSourceITCase.java (97%) rename hudi-flink/src/test/java/org/apache/hudi/{factory => table}/TestHoodieTableFactory.java (96%) rename hudi-flink/src/test/java/org/apache/hudi/{source => table}/TestHoodieTableSource.java (94%) rename hudi-flink/src/test/java/org/apache/hudi/{source => table}/format/TestInputFormat.java (96%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => }/utils/TestConfigurations.java (98%) rename hudi-flink/src/test/java/org/apache/hudi/{operator => }/utils/TestData.java (99%) diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/FlinkOptions.java b/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/operator/FlinkOptions.java rename to hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java index f725931c2..17eee61d8 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/FlinkOptions.java +++ b/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java @@ -16,14 +16,14 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.configuration; import org.apache.hudi.common.model.HoodieTableType; -import org.apache.hudi.config.HoodieWriteConfig; -import org.apache.hudi.streamer.FlinkStreamerConfig; import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload; +import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.keygen.SimpleAvroKeyGenerator; import org.apache.hudi.keygen.constant.KeyGeneratorOptions; +import org.apache.hudi.streamer.FlinkStreamerConfig; import org.apache.hudi.util.StreamerUtil; import org.apache.flink.configuration.ConfigOption; diff --git a/hudi-flink/src/main/java/org/apache/hudi/schema/FilebasedSchemaProvider.java b/hudi-flink/src/main/java/org/apache/hudi/schema/FilebasedSchemaProvider.java index 9a16c51bb..edcd928b2 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/schema/FilebasedSchemaProvider.java +++ b/hudi-flink/src/main/java/org/apache/hudi/schema/FilebasedSchemaProvider.java @@ -20,8 +20,8 @@ package org.apache.hudi.schema; import org.apache.hudi.common.config.TypedProperties; import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.exception.HoodieIOException; -import org.apache.hudi.operator.FlinkOptions; import org.apache.hudi.util.StreamerUtil; import org.apache.avro.Schema; diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/CommitSink.java b/hudi-flink/src/main/java/org/apache/hudi/sink/CommitSink.java index 3a12842e4..00ef66d2c 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/sink/CommitSink.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/CommitSink.java @@ -31,7 +31,6 @@ import org.apache.hudi.util.StreamerUtil; import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.functions.sink.RichSinkFunction; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/InstantGenerateOperator.java b/hudi-flink/src/main/java/org/apache/hudi/sink/InstantGenerateOperator.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/operator/InstantGenerateOperator.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/InstantGenerateOperator.java index d4e9e4d62..941379735 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/InstantGenerateOperator.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/InstantGenerateOperator.java @@ -16,9 +16,8 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.sink; -import org.apache.hudi.streamer.FlinkStreamerConfig; import org.apache.hudi.client.FlinkTaskContextSupplier; import org.apache.hudi.client.HoodieFlinkWriteClient; import org.apache.hudi.client.common.HoodieFlinkEngineContext; @@ -30,6 +29,8 @@ import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.common.table.timeline.HoodieTimeline; import org.apache.hudi.common.util.StringUtils; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.streamer.FlinkStreamerConfig; import org.apache.hudi.util.StreamerUtil; import org.apache.flink.api.common.state.ListState; @@ -40,11 +41,10 @@ import org.apache.flink.streaming.api.operators.AbstractStreamOperator; import org.apache.flink.streaming.api.operators.OneInputStreamOperator; import org.apache.flink.streaming.api.operators.StreamingRuntimeContext; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; -import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.PathFilter; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/KeyedWriteProcessFunction.java b/hudi-flink/src/main/java/org/apache/hudi/sink/KeyedWriteProcessFunction.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/operator/KeyedWriteProcessFunction.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/KeyedWriteProcessFunction.java index d7c0256b7..26ac820e0 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/KeyedWriteProcessFunction.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/KeyedWriteProcessFunction.java @@ -16,10 +16,8 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.sink; -import org.apache.hudi.index.HoodieIndex; -import org.apache.hudi.streamer.FlinkStreamerConfig; import org.apache.hudi.client.FlinkTaskContextSupplier; import org.apache.hudi.client.HoodieFlinkWriteClient; import org.apache.hudi.client.WriteStatus; @@ -29,6 +27,8 @@ import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.table.timeline.HoodieTimeline; import org.apache.hudi.exception.HoodieFlinkStreamerException; +import org.apache.hudi.index.HoodieIndex; +import org.apache.hudi.streamer.FlinkStreamerConfig; import org.apache.hudi.table.action.commit.FlinkWriteHelper; import org.apache.hudi.util.StreamerUtil; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/KeyedWriteProcessOperator.java b/hudi-flink/src/main/java/org/apache/hudi/sink/KeyedWriteProcessOperator.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/operator/KeyedWriteProcessOperator.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/KeyedWriteProcessOperator.java index 1d8e3c5ab..ebd15d169 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/KeyedWriteProcessOperator.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/KeyedWriteProcessOperator.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.sink; import org.apache.hudi.client.WriteStatus; import org.apache.hudi.common.model.HoodieRecord; @@ -26,7 +26,6 @@ import org.apache.flink.runtime.state.StateSnapshotContext; import org.apache.flink.streaming.api.functions.KeyedProcessFunction; import org.apache.flink.streaming.api.operators.KeyedProcessOperator; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/StreamWriteFunction.java b/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/operator/StreamWriteFunction.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java index d6657b04c..8f2269558 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/StreamWriteFunction.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.sink; import org.apache.hudi.client.FlinkTaskContextSupplier; import org.apache.hudi.client.HoodieFlinkWriteClient; @@ -26,8 +26,9 @@ import org.apache.hudi.common.config.SerializableConfiguration; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.WriteOperationType; import org.apache.hudi.common.util.ObjectSizeCalculator; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.index.HoodieIndex; -import org.apache.hudi.operator.event.BatchWriteSuccessEvent; +import org.apache.hudi.sink.event.BatchWriteSuccessEvent; import org.apache.hudi.table.action.commit.FlinkWriteHelper; import org.apache.hudi.util.StreamerUtil; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/StreamWriteOperator.java b/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperator.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/operator/StreamWriteOperator.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperator.java index 61585bc44..3150d065a 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/StreamWriteOperator.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperator.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.sink; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.operators.coordination.OperatorEvent; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/StreamWriteOperatorCoordinator.java b/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/operator/StreamWriteOperatorCoordinator.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java index 138fc550e..942c6a009 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/StreamWriteOperatorCoordinator.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java @@ -16,15 +16,16 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.sink; import org.apache.hudi.client.FlinkTaskContextSupplier; import org.apache.hudi.client.HoodieFlinkWriteClient; import org.apache.hudi.client.WriteStatus; import org.apache.hudi.client.common.HoodieFlinkEngineContext; import org.apache.hudi.common.util.Option; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.exception.HoodieException; -import org.apache.hudi.operator.event.BatchWriteSuccessEvent; +import org.apache.hudi.sink.event.BatchWriteSuccessEvent; import org.apache.hudi.util.StreamerUtil; import org.apache.flink.annotation.VisibleForTesting; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/StreamWriteOperatorFactory.java b/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorFactory.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/operator/StreamWriteOperatorFactory.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorFactory.java index 00d26a456..676390397 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/StreamWriteOperatorFactory.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorFactory.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.sink; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.jobgraph.OperatorID; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactFunction.java b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactFunction.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactFunction.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactFunction.java index 5b881fdd9..d0fa51baa 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactFunction.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactFunction.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.compact; +package org.apache.hudi.sink.compact; import org.apache.hudi.client.FlinkTaskContextSupplier; import org.apache.hudi.client.HoodieFlinkWriteClient; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactionCommitEvent.java b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitEvent.java similarity index 97% rename from hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactionCommitEvent.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitEvent.java index 3d77cf90f..744d32327 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactionCommitEvent.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitEvent.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.compact; +package org.apache.hudi.sink.compact; import org.apache.hudi.client.WriteStatus; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactionCommitSink.java b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitSink.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactionCommitSink.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitSink.java index f1ad7d070..da077a0e7 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactionCommitSink.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionCommitSink.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.compact; +package org.apache.hudi.sink.compact; import org.apache.hudi.avro.model.HoodieCompactionPlan; import org.apache.hudi.client.FlinkTaskContextSupplier; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactionPlanEvent.java b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanEvent.java similarity index 97% rename from hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactionPlanEvent.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanEvent.java index a19a2ee5d..c933192d9 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactionPlanEvent.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanEvent.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.compact; +package org.apache.hudi.sink.compact; import org.apache.hudi.common.model.CompactionOperation; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactionPlanOperator.java b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanOperator.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactionPlanOperator.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanOperator.java index b7b73731e..6eebe9445 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/compact/CompactionPlanOperator.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactionPlanOperator.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.compact; +package org.apache.hudi.sink.compact; import org.apache.hudi.avro.model.HoodieCompactionPlan; import org.apache.hudi.client.FlinkTaskContextSupplier; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/event/BatchWriteSuccessEvent.java b/hudi-flink/src/main/java/org/apache/hudi/sink/event/BatchWriteSuccessEvent.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/operator/event/BatchWriteSuccessEvent.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/event/BatchWriteSuccessEvent.java index f03e8d3e2..66521b101 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/event/BatchWriteSuccessEvent.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/event/BatchWriteSuccessEvent.java @@ -16,13 +16,13 @@ * limitations under the License. */ -package org.apache.hudi.operator.event; - -import org.apache.flink.runtime.operators.coordination.OperatorEvent; +package org.apache.hudi.sink.event; import org.apache.hudi.client.WriteStatus; import org.apache.hudi.common.util.ValidationUtils; +import org.apache.flink.runtime.operators.coordination.OperatorEvent; + import java.util.ArrayList; import java.util.List; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/BucketAssignFunction.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/BucketAssignFunction.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java index 9faf245ae..6bd30d2c4 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/BucketAssignFunction.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.partitioner; +package org.apache.hudi.sink.partitioner; import org.apache.hudi.client.FlinkTaskContextSupplier; import org.apache.hudi.client.common.HoodieFlinkEngineContext; @@ -30,10 +30,10 @@ import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.model.WriteOperationType; import org.apache.hudi.common.util.ParquetUtils; import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.exception.HoodieException; import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.index.HoodieIndexUtils; -import org.apache.hudi.operator.FlinkOptions; import org.apache.hudi.table.HoodieTable; import org.apache.hudi.table.action.commit.BucketInfo; import org.apache.hudi.util.StreamerUtil; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/BucketAssigner.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/BucketAssigner.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java index 54c9b4f47..58bbe9c4a 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/BucketAssigner.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigner.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.partitioner; +package org.apache.hudi.sink.partitioner; import org.apache.hudi.client.common.HoodieFlinkEngineContext; import org.apache.hudi.common.fs.FSUtils; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/BucketAssigners.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigners.java similarity index 93% rename from hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/BucketAssigners.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigners.java index b052c77ed..f5703f16f 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/BucketAssigners.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssigners.java @@ -16,12 +16,12 @@ * limitations under the License. */ -package org.apache.hudi.operator.partitioner; +package org.apache.hudi.sink.partitioner; import org.apache.hudi.client.common.HoodieFlinkEngineContext; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.config.HoodieWriteConfig; -import org.apache.hudi.operator.partitioner.delta.DeltaBucketAssigner; +import org.apache.hudi.sink.partitioner.delta.DeltaBucketAssigner; /** * Utilities for {@code BucketAssigner}. diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/delta/DeltaBucketAssigner.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/delta/DeltaBucketAssigner.java similarity index 97% rename from hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/delta/DeltaBucketAssigner.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/delta/DeltaBucketAssigner.java index 5b99cf3be..895f593fa 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/partitioner/delta/DeltaBucketAssigner.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/delta/DeltaBucketAssigner.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.partitioner.delta; +package org.apache.hudi.sink.partitioner.delta; import org.apache.hudi.client.common.HoodieFlinkEngineContext; import org.apache.hudi.common.fs.FSUtils; @@ -26,7 +26,7 @@ import org.apache.hudi.common.model.HoodieRecordLocation; import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.hudi.common.table.timeline.HoodieTimeline; import org.apache.hudi.config.HoodieWriteConfig; -import org.apache.hudi.operator.partitioner.BucketAssigner; +import org.apache.hudi.sink.partitioner.BucketAssigner; import org.apache.hudi.table.action.commit.SmallFile; import java.util.ArrayList; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/JsonStringToHoodieRecordMapFunction.java b/hudi-flink/src/main/java/org/apache/hudi/sink/transform/JsonStringToHoodieRecordMapFunction.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/source/JsonStringToHoodieRecordMapFunction.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/transform/JsonStringToHoodieRecordMapFunction.java index 8d0189cac..a14943cf9 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/JsonStringToHoodieRecordMapFunction.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/transform/JsonStringToHoodieRecordMapFunction.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.source; +package org.apache.hudi.sink.transform; import org.apache.hudi.avro.HoodieAvroUtils; import org.apache.hudi.common.config.TypedProperties; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/transform/RowDataToHoodieFunction.java b/hudi-flink/src/main/java/org/apache/hudi/sink/transform/RowDataToHoodieFunction.java similarity index 97% rename from hudi-flink/src/main/java/org/apache/hudi/operator/transform/RowDataToHoodieFunction.java rename to hudi-flink/src/main/java/org/apache/hudi/sink/transform/RowDataToHoodieFunction.java index 611a277d2..1d41003c5 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/transform/RowDataToHoodieFunction.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/transform/RowDataToHoodieFunction.java @@ -16,15 +16,15 @@ * limitations under the License. */ -package org.apache.hudi.operator.transform; +package org.apache.hudi.sink.transform; import org.apache.hudi.avro.HoodieAvroUtils; import org.apache.hudi.common.model.HoodieKey; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieRecordPayload; import org.apache.hudi.common.model.WriteOperationType; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.keygen.KeyGenerator; -import org.apache.hudi.operator.FlinkOptions; import org.apache.hudi.util.RowDataToAvroConverters; import org.apache.hudi.util.StreamerUtil; diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/StreamReadMonitoringFunction.java b/hudi-flink/src/main/java/org/apache/hudi/source/StreamReadMonitoringFunction.java similarity index 97% rename from hudi-flink/src/main/java/org/apache/hudi/operator/StreamReadMonitoringFunction.java rename to hudi-flink/src/main/java/org/apache/hudi/source/StreamReadMonitoringFunction.java index 2e24acaa9..d742c9e46 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/StreamReadMonitoringFunction.java +++ b/hudi-flink/src/main/java/org/apache/hudi/source/StreamReadMonitoringFunction.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.source; import org.apache.hudi.common.fs.FSUtils; import org.apache.hudi.common.model.HoodieCommitMetadata; @@ -27,9 +27,10 @@ import org.apache.hudi.common.table.timeline.HoodieTimeline; import org.apache.hudi.common.table.view.HoodieTableFileSystemView; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.ValidationUtils; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.exception.HoodieException; -import org.apache.hudi.source.format.mor.InstantRange; -import org.apache.hudi.source.format.mor.MergeOnReadInputSplit; +import org.apache.hudi.table.format.mor.InstantRange; +import org.apache.hudi.table.format.mor.MergeOnReadInputSplit; import org.apache.hudi.util.StreamerUtil; import org.apache.flink.annotation.VisibleForTesting; @@ -341,13 +342,13 @@ public class StreamReadMonitoringFunction * @return the file statuses array */ private FileStatus[] getWritePathsOfInstants(List metadataList) { - FileSystem fs = FSUtils.getFs(path.getPath(), hadoopConf); + FileSystem fs = FSUtils.getFs(path.toString(), hadoopConf); return metadataList.stream().map(metadata -> getWritePathsOfInstant(metadata, fs)) .flatMap(Collection::stream).toArray(FileStatus[]::new); } private List getWritePathsOfInstant(HoodieCommitMetadata metadata, FileSystem fs) { - return metadata.getFileIdAndFullPaths(path.getPath()).values().stream() + return metadata.getFileIdAndFullPaths(path.toString()).values().stream() .map(path -> { try { return fs.getFileStatus(new org.apache.hadoop.fs.Path(path)); diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/StreamReadOperator.java b/hudi-flink/src/main/java/org/apache/hudi/source/StreamReadOperator.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/operator/StreamReadOperator.java rename to hudi-flink/src/main/java/org/apache/hudi/source/StreamReadOperator.java index d12147d28..b974c7ff3 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/operator/StreamReadOperator.java +++ b/hudi-flink/src/main/java/org/apache/hudi/source/StreamReadOperator.java @@ -16,10 +16,10 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.source; -import org.apache.hudi.source.format.mor.MergeOnReadInputFormat; -import org.apache.hudi.source.format.mor.MergeOnReadInputSplit; +import org.apache.hudi.table.format.mor.MergeOnReadInputFormat; +import org.apache.hudi.table.format.mor.MergeOnReadInputSplit; import org.apache.flink.api.common.state.ListState; import org.apache.flink.api.common.state.ListStateDescriptor; diff --git a/hudi-flink/src/main/java/org/apache/hudi/streamer/HoodieFlinkStreamer.java b/hudi-flink/src/main/java/org/apache/hudi/streamer/HoodieFlinkStreamer.java index 29f5de8c2..0d37574a4 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/streamer/HoodieFlinkStreamer.java +++ b/hudi-flink/src/main/java/org/apache/hudi/streamer/HoodieFlinkStreamer.java @@ -22,13 +22,13 @@ import org.apache.hudi.client.WriteStatus; import org.apache.hudi.common.config.TypedProperties; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.config.HoodieWriteConfig; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.operator.InstantGenerateOperator; -import org.apache.hudi.operator.KeyedWriteProcessFunction; -import org.apache.hudi.operator.KeyedWriteProcessOperator; -import org.apache.hudi.operator.partitioner.BucketAssignFunction; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.sink.CommitSink; -import org.apache.hudi.source.JsonStringToHoodieRecordMapFunction; +import org.apache.hudi.sink.InstantGenerateOperator; +import org.apache.hudi.sink.KeyedWriteProcessFunction; +import org.apache.hudi.sink.KeyedWriteProcessOperator; +import org.apache.hudi.sink.partitioner.BucketAssignFunction; +import org.apache.hudi.sink.transform.JsonStringToHoodieRecordMapFunction; import org.apache.hudi.util.StreamerUtil; import com.beust.jcommander.JCommander; diff --git a/hudi-flink/src/main/java/org/apache/hudi/streamer/HoodieFlinkStreamerV2.java b/hudi-flink/src/main/java/org/apache/hudi/streamer/HoodieFlinkStreamerV2.java index d217e08f9..c62307e3b 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/streamer/HoodieFlinkStreamerV2.java +++ b/hudi-flink/src/main/java/org/apache/hudi/streamer/HoodieFlinkStreamerV2.java @@ -19,10 +19,10 @@ package org.apache.hudi.streamer; import org.apache.hudi.common.model.HoodieRecord; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.operator.StreamWriteOperatorFactory; -import org.apache.hudi.operator.partitioner.BucketAssignFunction; -import org.apache.hudi.operator.transform.RowDataToHoodieFunction; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.sink.StreamWriteOperatorFactory; +import org.apache.hudi.sink.partitioner.BucketAssignFunction; +import org.apache.hudi.sink.transform.RowDataToHoodieFunction; import org.apache.hudi.util.AvroSchemaConverter; import org.apache.hudi.util.StreamerUtil; diff --git a/hudi-flink/src/main/java/org/apache/hudi/factory/HoodieTableFactory.java b/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java similarity index 96% rename from hudi-flink/src/main/java/org/apache/hudi/factory/HoodieTableFactory.java rename to hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java index 28883ba04..b16cfdb6e 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/factory/HoodieTableFactory.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java @@ -16,11 +16,9 @@ * limitations under the License. */ -package org.apache.hudi.factory; +package org.apache.hudi.table; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.sink.HoodieTableSink; -import org.apache.hudi.source.HoodieTableSource; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.util.AvroSchemaConverter; import org.apache.flink.configuration.Configuration; diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/HoodieTableSink.java b/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java similarity index 89% rename from hudi-flink/src/main/java/org/apache/hudi/sink/HoodieTableSink.java rename to hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java index fde156043..c973f0865 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/sink/HoodieTableSink.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSink.java @@ -16,18 +16,18 @@ * limitations under the License. */ -package org.apache.hudi.sink; +package org.apache.hudi.table; import org.apache.hudi.common.model.HoodieRecord; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.operator.StreamWriteOperatorFactory; -import org.apache.hudi.operator.compact.CompactFunction; -import org.apache.hudi.operator.compact.CompactionCommitEvent; -import org.apache.hudi.operator.compact.CompactionCommitSink; -import org.apache.hudi.operator.compact.CompactionPlanEvent; -import org.apache.hudi.operator.compact.CompactionPlanOperator; -import org.apache.hudi.operator.partitioner.BucketAssignFunction; -import org.apache.hudi.operator.transform.RowDataToHoodieFunction; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.sink.StreamWriteOperatorFactory; +import org.apache.hudi.sink.compact.CompactFunction; +import org.apache.hudi.sink.compact.CompactionCommitEvent; +import org.apache.hudi.sink.compact.CompactionCommitSink; +import org.apache.hudi.sink.compact.CompactionPlanEvent; +import org.apache.hudi.sink.compact.CompactionPlanOperator; +import org.apache.hudi.sink.partitioner.BucketAssignFunction; +import org.apache.hudi.sink.transform.RowDataToHoodieFunction; import org.apache.hudi.util.StreamerUtil; import org.apache.flink.annotation.VisibleForTesting; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/HoodieTableSource.java b/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java similarity index 96% rename from hudi-flink/src/main/java/org/apache/hudi/source/HoodieTableSource.java rename to hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java index e78c65613..4978ed5e0 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/HoodieTableSource.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableSource.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.source; +package org.apache.hudi.table; import org.apache.hudi.common.fs.FSUtils; import org.apache.hudi.common.model.HoodieBaseFile; @@ -27,17 +27,17 @@ import org.apache.hudi.common.table.TableSchemaResolver; import org.apache.hudi.common.table.view.HoodieTableFileSystemView; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.ValidationUtils; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.exception.HoodieException; import org.apache.hudi.hadoop.HoodieROTablePathFilter; import org.apache.hudi.hadoop.utils.HoodieRealtimeInputFormatUtils; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.operator.StreamReadMonitoringFunction; -import org.apache.hudi.operator.StreamReadOperator; -import org.apache.hudi.source.format.FilePathUtils; -import org.apache.hudi.source.format.cow.CopyOnWriteInputFormat; -import org.apache.hudi.source.format.mor.MergeOnReadInputFormat; -import org.apache.hudi.source.format.mor.MergeOnReadInputSplit; -import org.apache.hudi.source.format.mor.MergeOnReadTableState; +import org.apache.hudi.source.StreamReadMonitoringFunction; +import org.apache.hudi.source.StreamReadOperator; +import org.apache.hudi.table.format.FilePathUtils; +import org.apache.hudi.table.format.cow.CopyOnWriteInputFormat; +import org.apache.hudi.table.format.mor.MergeOnReadInputFormat; +import org.apache.hudi.table.format.mor.MergeOnReadInputSplit; +import org.apache.hudi.table.format.mor.MergeOnReadTableState; import org.apache.hudi.util.AvroSchemaConverter; import org.apache.hudi.util.StreamerUtil; @@ -87,7 +87,7 @@ import java.util.stream.Collectors; import java.util.stream.IntStream; import static org.apache.hudi.hadoop.utils.HoodieRealtimeRecordReaderUtils.getMaxCompactionMemoryInBytes; -import static org.apache.hudi.source.format.FormatUtils.getParquetConf; +import static org.apache.hudi.table.format.FormatUtils.getParquetConf; /** * Hoodie batch table source that always read the latest snapshot of the underneath table. diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/FilePathUtils.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/FilePathUtils.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/FilePathUtils.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/FilePathUtils.java index 1b029b225..4dfe99036 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/FilePathUtils.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/FilePathUtils.java @@ -16,10 +16,10 @@ * limitations under the License. */ -package org.apache.hudi.source.format; +package org.apache.hudi.table.format; import org.apache.hudi.common.fs.FSUtils; -import org.apache.hudi.operator.FlinkOptions; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.table.api.TableException; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/FormatUtils.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java similarity index 96% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/FormatUtils.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java index c2192f428..702614b29 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/FormatUtils.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/FormatUtils.java @@ -16,13 +16,13 @@ * limitations under the License. */ -package org.apache.hudi.source.format; +package org.apache.hudi.table.format; import org.apache.hudi.common.fs.FSUtils; import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.hadoop.config.HoodieRealtimeConfig; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.source.format.mor.MergeOnReadInputSplit; +import org.apache.hudi.table.format.mor.MergeOnReadInputSplit; import org.apache.avro.Schema; import org.apache.avro.generic.GenericRecord; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/AbstractColumnReader.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/AbstractColumnReader.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/cow/AbstractColumnReader.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/AbstractColumnReader.java index 4f83be77f..e6f40a557 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/AbstractColumnReader.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/AbstractColumnReader.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.source.format.cow; +package org.apache.hudi.table.format.cow; import org.apache.flink.formats.parquet.vector.ParquetDictionary; import org.apache.flink.formats.parquet.vector.reader.ColumnReader; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/CopyOnWriteInputFormat.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/cow/CopyOnWriteInputFormat.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java index 709c32e1d..77f3e2a10 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/CopyOnWriteInputFormat.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.source.format.cow; +package org.apache.hudi.table.format.cow; import org.apache.hudi.common.fs.FSUtils; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/Int64TimestampColumnReader.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/Int64TimestampColumnReader.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/cow/Int64TimestampColumnReader.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/Int64TimestampColumnReader.java index df779de41..f71f6e599 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/Int64TimestampColumnReader.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/Int64TimestampColumnReader.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.hudi.source.format.cow; +package org.apache.hudi.table.format.cow; import org.apache.flink.table.data.TimestampData; import org.apache.flink.table.data.vector.writable.WritableIntVector; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetColumnarRowSplitReader.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/ParquetColumnarRowSplitReader.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetColumnarRowSplitReader.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/ParquetColumnarRowSplitReader.java index d163f30dd..415469695 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetColumnarRowSplitReader.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/ParquetColumnarRowSplitReader.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.source.format.cow; +package org.apache.hudi.table.format.cow; import org.apache.flink.formats.parquet.vector.reader.ColumnReader; import org.apache.flink.table.data.ColumnarRowData; @@ -47,8 +47,8 @@ import java.util.List; import java.util.Locale; import java.util.Map; -import static org.apache.hudi.source.format.cow.ParquetSplitReaderUtil.createColumnReader; -import static org.apache.hudi.source.format.cow.ParquetSplitReaderUtil.createWritableColumnVector; +import static org.apache.hudi.table.format.cow.ParquetSplitReaderUtil.createColumnReader; +import static org.apache.hudi.table.format.cow.ParquetSplitReaderUtil.createWritableColumnVector; import static org.apache.parquet.filter2.compat.RowGroupFilter.filterRowGroups; import static org.apache.parquet.format.converter.ParquetMetadataConverter.range; import static org.apache.parquet.hadoop.ParquetFileReader.readFooter; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetDecimalVector.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/ParquetDecimalVector.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetDecimalVector.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/ParquetDecimalVector.java index feaa65746..2bf55b35d 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetDecimalVector.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/ParquetDecimalVector.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.source.format.cow; +package org.apache.hudi.table.format.cow; import org.apache.flink.table.data.DecimalData; import org.apache.flink.table.data.DecimalDataUtils; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetSplitReaderUtil.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetSplitReaderUtil.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java index a047c7e6c..b626e8833 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/ParquetSplitReaderUtil.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.source.format.cow; +package org.apache.hudi.table.format.cow; import org.apache.flink.core.fs.Path; import org.apache.flink.formats.parquet.vector.reader.BooleanColumnReader; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/RunLengthDecoder.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/RunLengthDecoder.java similarity index 99% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/cow/RunLengthDecoder.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/cow/RunLengthDecoder.java index d9fc85560..159574714 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/cow/RunLengthDecoder.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/RunLengthDecoder.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.source.format.cow; +package org.apache.hudi.table.format.cow; import org.apache.flink.table.data.vector.writable.WritableColumnVector; import org.apache.flink.table.data.vector.writable.WritableIntVector; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/mor/InstantRange.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/InstantRange.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/mor/InstantRange.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/mor/InstantRange.java index 62f34db9f..a90d4f430 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/mor/InstantRange.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/InstantRange.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.source.format.mor; +package org.apache.hudi.table.format.mor; import org.apache.hudi.common.table.timeline.HoodieTimeline; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/mor/MergeOnReadInputFormat.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java similarity index 97% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/mor/MergeOnReadInputFormat.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java index e2951ab49..a3a6f858a 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/mor/MergeOnReadInputFormat.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputFormat.java @@ -16,17 +16,17 @@ * limitations under the License. */ -package org.apache.hudi.source.format.mor; +package org.apache.hudi.table.format.mor; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieRecordPayload; import org.apache.hudi.common.util.Option; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.exception.HoodieException; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.source.format.FilePathUtils; -import org.apache.hudi.source.format.FormatUtils; -import org.apache.hudi.source.format.cow.ParquetColumnarRowSplitReader; -import org.apache.hudi.source.format.cow.ParquetSplitReaderUtil; +import org.apache.hudi.table.format.FilePathUtils; +import org.apache.hudi.table.format.FormatUtils; +import org.apache.hudi.table.format.cow.ParquetColumnarRowSplitReader; +import org.apache.hudi.table.format.cow.ParquetSplitReaderUtil; import org.apache.hudi.util.AvroToRowDataConverters; import org.apache.hudi.util.RowDataToAvroConverters; import org.apache.hudi.util.StreamerUtil; @@ -57,7 +57,7 @@ import java.util.stream.IntStream; import static org.apache.flink.table.data.vector.VectorizedColumnBatch.DEFAULT_SIZE; import static org.apache.flink.table.filesystem.RowPartitionComputer.restorePartValueFromType; import static org.apache.hudi.hadoop.utils.HoodieInputFormatUtils.HOODIE_RECORD_KEY_COL_POS; -import static org.apache.hudi.source.format.FormatUtils.buildAvroRecordBySchema; +import static org.apache.hudi.table.format.FormatUtils.buildAvroRecordBySchema; /** * The base InputFormat class to read from Hoodie data + log files. diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/mor/MergeOnReadInputSplit.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputSplit.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/mor/MergeOnReadInputSplit.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputSplit.java index a73e93a48..7aaa0ad8e 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/mor/MergeOnReadInputSplit.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadInputSplit.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.source.format.mor; +package org.apache.hudi.table.format.mor; import org.apache.hudi.common.util.Option; diff --git a/hudi-flink/src/main/java/org/apache/hudi/source/format/mor/MergeOnReadTableState.java b/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadTableState.java similarity index 98% rename from hudi-flink/src/main/java/org/apache/hudi/source/format/mor/MergeOnReadTableState.java rename to hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadTableState.java index 6b90352f5..7dedcdaa3 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/source/format/mor/MergeOnReadTableState.java +++ b/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadTableState.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.source.format.mor; +package org.apache.hudi.table.format.mor; import org.apache.flink.table.types.logical.RowType; diff --git a/hudi-flink/src/main/java/org/apache/hudi/util/RowDataToAvroConverters.java b/hudi-flink/src/main/java/org/apache/hudi/util/RowDataToAvroConverters.java index d5ff66388..3a531c096 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/util/RowDataToAvroConverters.java +++ b/hudi-flink/src/main/java/org/apache/hudi/util/RowDataToAvroConverters.java @@ -18,6 +18,10 @@ package org.apache.hudi.util; +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericData; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.util.Utf8; import org.apache.flink.annotation.Internal; import org.apache.flink.table.data.ArrayData; import org.apache.flink.table.data.DecimalData; @@ -28,11 +32,6 @@ import org.apache.flink.table.types.logical.ArrayType; import org.apache.flink.table.types.logical.LogicalType; import org.apache.flink.table.types.logical.RowType; -import org.apache.avro.Schema; -import org.apache.avro.generic.GenericData; -import org.apache.avro.generic.GenericRecord; -import org.apache.avro.util.Utf8; - import java.io.Serializable; import java.nio.ByteBuffer; import java.util.ArrayList; diff --git a/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java b/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java index 536423060..005860365 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java +++ b/hudi-flink/src/main/java/org/apache/hudi/util/StreamerUtil.java @@ -30,12 +30,12 @@ import org.apache.hudi.common.util.ReflectionUtils; import org.apache.hudi.common.util.TablePathUtils; import org.apache.hudi.config.HoodieCompactionConfig; import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.exception.HoodieException; import org.apache.hudi.exception.HoodieIOException; import org.apache.hudi.exception.TableNotFoundException; import org.apache.hudi.keygen.KeyGenerator; import org.apache.hudi.keygen.SimpleAvroKeyGenerator; -import org.apache.hudi.operator.FlinkOptions; import org.apache.hudi.schema.FilebasedSchemaProvider; import org.apache.hudi.streamer.FlinkStreamerConfig; import org.apache.hudi.table.action.compact.CompactionTriggerStrategy; diff --git a/hudi-flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory b/hudi-flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory index 105f64cfb..1031e4274 100644 --- a/hudi-flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory +++ b/hudi-flink/src/main/resources/META-INF/services/org.apache.flink.table.factories.TableFactory @@ -14,4 +14,4 @@ # See the License for the specific language governing permissions and # limitations under the License. -org.apache.hudi.factory.HoodieTableFactory +org.apache.hudi.table.HoodieTableFactory diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/StreamWriteITCase.java b/hudi-flink/src/test/java/org/apache/hudi/sink/StreamWriteITCase.java similarity index 95% rename from hudi-flink/src/test/java/org/apache/hudi/operator/StreamWriteITCase.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/StreamWriteITCase.java index 6e1fe475d..a446ee469 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/StreamWriteITCase.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/StreamWriteITCase.java @@ -16,24 +16,24 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.sink; import org.apache.hudi.client.WriteStatus; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieTableType; -import org.apache.hudi.operator.compact.CompactFunction; -import org.apache.hudi.operator.compact.CompactionCommitEvent; -import org.apache.hudi.operator.compact.CompactionCommitSink; -import org.apache.hudi.operator.compact.CompactionPlanEvent; -import org.apache.hudi.operator.compact.CompactionPlanOperator; -import org.apache.hudi.operator.partitioner.BucketAssignFunction; -import org.apache.hudi.operator.transform.RowDataToHoodieFunction; -import org.apache.hudi.operator.utils.TestConfigurations; -import org.apache.hudi.operator.utils.TestData; -import org.apache.hudi.sink.CommitSink; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.sink.compact.CompactFunction; +import org.apache.hudi.sink.compact.CompactionCommitEvent; +import org.apache.hudi.sink.compact.CompactionCommitSink; +import org.apache.hudi.sink.compact.CompactionPlanEvent; +import org.apache.hudi.sink.compact.CompactionPlanOperator; +import org.apache.hudi.sink.partitioner.BucketAssignFunction; +import org.apache.hudi.sink.transform.RowDataToHoodieFunction; import org.apache.hudi.streamer.FlinkStreamerConfig; import org.apache.hudi.util.AvroSchemaConverter; import org.apache.hudi.util.StreamerUtil; +import org.apache.hudi.utils.TestConfigurations; +import org.apache.hudi.utils.TestData; import org.apache.hudi.utils.source.ContinuousFileSource; import org.apache.flink.api.common.JobStatus; diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/StreamWriteOperatorCoordinatorTest.java b/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java similarity index 96% rename from hudi-flink/src/test/java/org/apache/hudi/operator/StreamWriteOperatorCoordinatorTest.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java index aa8be8e4c..b68f62e89 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/StreamWriteOperatorCoordinatorTest.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java @@ -16,16 +16,16 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.sink; import org.apache.hudi.client.WriteStatus; import org.apache.hudi.common.fs.FSUtils; import org.apache.hudi.common.model.HoodieWriteStat; import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.exception.HoodieException; -import org.apache.hudi.operator.event.BatchWriteSuccessEvent; -import org.apache.hudi.operator.utils.TestConfigurations; +import org.apache.hudi.sink.event.BatchWriteSuccessEvent; import org.apache.hudi.util.StreamerUtil; +import org.apache.hudi.utils.TestConfigurations; import org.apache.flink.runtime.operators.coordination.OperatorEvent; import org.apache.hadoop.fs.FileSystem; @@ -49,7 +49,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; /** * Test cases for StreamingSinkOperatorCoordinator. */ -public class StreamWriteOperatorCoordinatorTest { +public class TestStreamWriteOperatorCoordinator { private StreamWriteOperatorCoordinator coordinator; @TempDir diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/TestWriteCopyOnWrite.java b/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java similarity index 98% rename from hudi-flink/src/test/java/org/apache/hudi/operator/TestWriteCopyOnWrite.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java index b8c9b3c22..1167779ff 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/TestWriteCopyOnWrite.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteCopyOnWrite.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.sink; import org.apache.hudi.client.HoodieFlinkWriteClient; import org.apache.hudi.client.WriteStatus; @@ -24,11 +24,12 @@ import org.apache.hudi.common.model.HoodieKey; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.exception.HoodieException; -import org.apache.hudi.operator.event.BatchWriteSuccessEvent; -import org.apache.hudi.operator.utils.StreamWriteFunctionWrapper; -import org.apache.hudi.operator.utils.TestConfigurations; -import org.apache.hudi.operator.utils.TestData; +import org.apache.hudi.sink.event.BatchWriteSuccessEvent; +import org.apache.hudi.sink.utils.StreamWriteFunctionWrapper; +import org.apache.hudi.utils.TestConfigurations; +import org.apache.hudi.utils.TestData; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.operators.coordination.OperatorEvent; diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/TestWriteMergeOnRead.java b/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java similarity index 97% rename from hudi-flink/src/test/java/org/apache/hudi/operator/TestWriteMergeOnRead.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java index 0b305a7c1..5ce8ae2ed 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/TestWriteMergeOnRead.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnRead.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.sink; import org.apache.hudi.client.FlinkTaskContextSupplier; import org.apache.hudi.client.common.HoodieFlinkEngineContext; @@ -28,9 +28,9 @@ import org.apache.hudi.common.table.TableSchemaResolver; import org.apache.hudi.common.table.timeline.HoodieActiveTimeline; import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.hudi.config.HoodieWriteConfig; -import org.apache.hudi.operator.utils.TestData; import org.apache.hudi.table.HoodieFlinkTable; import org.apache.hudi.util.StreamerUtil; +import org.apache.hudi.utils.TestData; import org.apache.avro.Schema; import org.apache.hadoop.fs.FileSystem; diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/TestWriteMergeOnReadWithCompact.java b/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnReadWithCompact.java similarity index 95% rename from hudi-flink/src/test/java/org/apache/hudi/operator/TestWriteMergeOnReadWithCompact.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnReadWithCompact.java index 9a9312c83..2821caeef 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/TestWriteMergeOnReadWithCompact.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/TestWriteMergeOnReadWithCompact.java @@ -16,9 +16,10 @@ * limitations under the License. */ -package org.apache.hudi.operator; +package org.apache.hudi.sink; import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.flink.configuration.Configuration; import org.junit.jupiter.api.Disabled; diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/partitioner/TestBucketAssigner.java b/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/TestBucketAssigner.java similarity index 98% rename from hudi-flink/src/test/java/org/apache/hudi/operator/partitioner/TestBucketAssigner.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/TestBucketAssigner.java index e27ea0757..04cee442a 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/partitioner/TestBucketAssigner.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/partitioner/TestBucketAssigner.java @@ -16,18 +16,18 @@ * limitations under the License. */ -package org.apache.hudi.operator.partitioner; +package org.apache.hudi.sink.partitioner; import org.apache.hudi.client.FlinkTaskContextSupplier; import org.apache.hudi.client.common.HoodieFlinkEngineContext; import org.apache.hudi.common.config.SerializableConfiguration; import org.apache.hudi.common.model.HoodieRecordLocation; import org.apache.hudi.config.HoodieWriteConfig; -import org.apache.hudi.operator.utils.TestConfigurations; import org.apache.hudi.table.action.commit.BucketInfo; import org.apache.hudi.table.action.commit.BucketType; import org.apache.hudi.table.action.commit.SmallFile; import org.apache.hudi.util.StreamerUtil; +import org.apache.hudi.utils.TestConfigurations; import org.apache.flink.configuration.Configuration; import org.junit.jupiter.api.BeforeEach; diff --git a/hudi-flink/src/test/java/org/apache/hudi/source/TestJsonStringToHoodieRecordMapFunction.java b/hudi-flink/src/test/java/org/apache/hudi/sink/transform/TestJsonStringToHoodieRecordMapFunction.java similarity index 98% rename from hudi-flink/src/test/java/org/apache/hudi/source/TestJsonStringToHoodieRecordMapFunction.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/transform/TestJsonStringToHoodieRecordMapFunction.java index 58eefc994..c2d7bdc5f 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/source/TestJsonStringToHoodieRecordMapFunction.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/transform/TestJsonStringToHoodieRecordMapFunction.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.source; +package org.apache.hudi.sink.transform; import org.apache.hudi.common.config.TypedProperties; import org.apache.hudi.common.model.HoodieRecord; diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/CompactFunctionWrapper.java b/hudi-flink/src/test/java/org/apache/hudi/sink/utils/CompactFunctionWrapper.java similarity index 91% rename from hudi-flink/src/test/java/org/apache/hudi/operator/utils/CompactFunctionWrapper.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/utils/CompactFunctionWrapper.java index e2fcb04f1..e8796ec72 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/CompactFunctionWrapper.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/utils/CompactFunctionWrapper.java @@ -16,14 +16,14 @@ * limitations under the License. */ -package org.apache.hudi.operator.utils; +package org.apache.hudi.sink.utils; import org.apache.hudi.avro.model.HoodieCompactionPlan; -import org.apache.hudi.operator.compact.CompactFunction; -import org.apache.hudi.operator.compact.CompactionCommitEvent; -import org.apache.hudi.operator.compact.CompactionCommitSink; -import org.apache.hudi.operator.compact.CompactionPlanEvent; -import org.apache.hudi.operator.compact.CompactionPlanOperator; +import org.apache.hudi.sink.compact.CompactFunction; +import org.apache.hudi.sink.compact.CompactionCommitEvent; +import org.apache.hudi.sink.compact.CompactionCommitSink; +import org.apache.hudi.sink.compact.CompactionPlanEvent; +import org.apache.hudi.sink.compact.CompactionPlanOperator; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.io.disk.iomanager.IOManager; @@ -43,7 +43,7 @@ import java.util.ArrayList; import java.util.List; /** - * A wrapper class to manipulate the {@link org.apache.hudi.operator.compact.CompactFunction} instance for testing. + * A wrapper class to manipulate the {@link org.apache.hudi.sink.compact.CompactFunction} instance for testing. */ public class CompactFunctionWrapper { private final Configuration conf; diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/MockFunctionInitializationContext.java b/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockFunctionInitializationContext.java similarity index 97% rename from hudi-flink/src/test/java/org/apache/hudi/operator/utils/MockFunctionInitializationContext.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockFunctionInitializationContext.java index 2aae03326..e3da4c942 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/MockFunctionInitializationContext.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockFunctionInitializationContext.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.utils; +package org.apache.hudi.sink.utils; import org.apache.flink.api.common.state.KeyedStateStore; import org.apache.flink.runtime.state.FunctionInitializationContext; diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/MockMapState.java b/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockMapState.java similarity index 98% rename from hudi-flink/src/test/java/org/apache/hudi/operator/utils/MockMapState.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockMapState.java index 96972986b..b0d1c285d 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/MockMapState.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockMapState.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.utils; +package org.apache.hudi.sink.utils; import org.apache.flink.api.common.state.MapState; diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/MockOperatorStateStore.java b/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockOperatorStateStore.java similarity index 99% rename from hudi-flink/src/test/java/org/apache/hudi/operator/utils/MockOperatorStateStore.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockOperatorStateStore.java index 016ad5bec..192183eff 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/MockOperatorStateStore.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockOperatorStateStore.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.utils; +package org.apache.hudi.sink.utils; import org.apache.flink.api.common.state.AggregatingState; import org.apache.flink.api.common.state.AggregatingStateDescriptor; diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/MockStreamingRuntimeContext.java b/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockStreamingRuntimeContext.java similarity index 99% rename from hudi-flink/src/test/java/org/apache/hudi/operator/utils/MockStreamingRuntimeContext.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockStreamingRuntimeContext.java index 1db98df86..14305da3d 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/MockStreamingRuntimeContext.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/utils/MockStreamingRuntimeContext.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.utils; +package org.apache.hudi.sink.utils; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.state.KeyedStateStore; diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/StreamWriteFunctionWrapper.java b/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java similarity index 94% rename from hudi-flink/src/test/java/org/apache/hudi/operator/utils/StreamWriteFunctionWrapper.java rename to hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java index 876684489..ee7774d56 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/StreamWriteFunctionWrapper.java +++ b/hudi-flink/src/test/java/org/apache/hudi/sink/utils/StreamWriteFunctionWrapper.java @@ -16,18 +16,19 @@ * limitations under the License. */ -package org.apache.hudi.operator.utils; +package org.apache.hudi.sink.utils; import org.apache.hudi.client.HoodieFlinkWriteClient; import org.apache.hudi.common.model.HoodieKey; import org.apache.hudi.common.model.HoodieRecord; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.exception.HoodieException; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.operator.StreamWriteFunction; -import org.apache.hudi.operator.StreamWriteOperatorCoordinator; -import org.apache.hudi.operator.event.BatchWriteSuccessEvent; -import org.apache.hudi.operator.partitioner.BucketAssignFunction; -import org.apache.hudi.operator.transform.RowDataToHoodieFunction; +import org.apache.hudi.sink.StreamWriteFunction; +import org.apache.hudi.sink.StreamWriteOperatorCoordinator; +import org.apache.hudi.sink.event.BatchWriteSuccessEvent; +import org.apache.hudi.sink.partitioner.BucketAssignFunction; +import org.apache.hudi.sink.transform.RowDataToHoodieFunction; +import org.apache.hudi.utils.TestConfigurations; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.io.disk.iomanager.IOManager; diff --git a/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadMonitoringFunction.java b/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadMonitoringFunction.java index 4733fa07b..406da3228 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadMonitoringFunction.java +++ b/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadMonitoringFunction.java @@ -18,12 +18,11 @@ package org.apache.hudi.source; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.operator.StreamReadMonitoringFunction; -import org.apache.hudi.operator.utils.TestConfigurations; -import org.apache.hudi.operator.utils.TestData; -import org.apache.hudi.source.format.mor.MergeOnReadInputSplit; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.table.format.mor.MergeOnReadInputSplit; import org.apache.hudi.util.StreamerUtil; +import org.apache.hudi.utils.TestConfigurations; +import org.apache.hudi.utils.TestData; import org.apache.hudi.utils.TestUtils; import org.apache.flink.configuration.Configuration; diff --git a/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadOperator.java b/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadOperator.java index 4ada381ba..aaeed94e2 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadOperator.java +++ b/hudi-flink/src/test/java/org/apache/hudi/source/TestStreamReadOperator.java @@ -20,18 +20,16 @@ package org.apache.hudi.source; import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.common.table.TableSchemaResolver; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.exception.HoodieException; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.operator.StreamReadMonitoringFunction; -import org.apache.hudi.operator.StreamReadOperator; -import org.apache.hudi.operator.utils.TestConfigurations; -import org.apache.hudi.operator.utils.TestData; -import org.apache.hudi.source.format.FilePathUtils; -import org.apache.hudi.source.format.mor.MergeOnReadInputFormat; -import org.apache.hudi.source.format.mor.MergeOnReadInputSplit; -import org.apache.hudi.source.format.mor.MergeOnReadTableState; +import org.apache.hudi.table.format.FilePathUtils; +import org.apache.hudi.table.format.mor.MergeOnReadInputFormat; +import org.apache.hudi.table.format.mor.MergeOnReadInputSplit; +import org.apache.hudi.table.format.mor.MergeOnReadTableState; import org.apache.hudi.util.AvroSchemaConverter; import org.apache.hudi.util.StreamerUtil; +import org.apache.hudi.utils.TestConfigurations; +import org.apache.hudi.utils.TestData; import org.apache.hudi.utils.TestUtils; import org.apache.avro.Schema; diff --git a/hudi-flink/src/test/java/org/apache/hudi/source/HoodieDataSourceITCase.java b/hudi-flink/src/test/java/org/apache/hudi/table/HoodieDataSourceITCase.java similarity index 97% rename from hudi-flink/src/test/java/org/apache/hudi/source/HoodieDataSourceITCase.java rename to hudi-flink/src/test/java/org/apache/hudi/table/HoodieDataSourceITCase.java index e3138af9e..eca14e9a2 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/source/HoodieDataSourceITCase.java +++ b/hudi-flink/src/test/java/org/apache/hudi/table/HoodieDataSourceITCase.java @@ -16,11 +16,11 @@ * limitations under the License. */ -package org.apache.hudi.source; +package org.apache.hudi.table; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.operator.utils.TestConfigurations; -import org.apache.hudi.operator.utils.TestData; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.utils.TestConfigurations; +import org.apache.hudi.utils.TestData; import org.apache.hudi.utils.TestUtils; import org.apache.hudi.utils.factory.CollectSinkTableFactory; @@ -46,7 +46,7 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; -import static org.apache.hudi.operator.utils.TestData.assertRowsEquals; +import static org.apache.hudi.utils.TestData.assertRowsEquals; /** * IT cases for Hoodie table source and sink. diff --git a/hudi-flink/src/test/java/org/apache/hudi/factory/TestHoodieTableFactory.java b/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java similarity index 96% rename from hudi-flink/src/test/java/org/apache/hudi/factory/TestHoodieTableFactory.java rename to hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java index f2d4ea6c4..f64808e95 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/factory/TestHoodieTableFactory.java +++ b/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableFactory.java @@ -16,13 +16,11 @@ * limitations under the License. */ -package org.apache.hudi.factory; +package org.apache.hudi.table; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.operator.utils.TestConfigurations; -import org.apache.hudi.sink.HoodieTableSink; -import org.apache.hudi.source.HoodieTableSource; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.util.StreamerUtil; +import org.apache.hudi.utils.TestConfigurations; import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.ReadableConfig; diff --git a/hudi-flink/src/test/java/org/apache/hudi/source/TestHoodieTableSource.java b/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java similarity index 94% rename from hudi-flink/src/test/java/org/apache/hudi/source/TestHoodieTableSource.java rename to hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java index c1e234855..2acc3234e 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/source/TestHoodieTableSource.java +++ b/hudi-flink/src/test/java/org/apache/hudi/table/TestHoodieTableSource.java @@ -16,14 +16,14 @@ * limitations under the License. */ -package org.apache.hudi.source; +package org.apache.hudi.table; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.exception.HoodieException; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.operator.utils.TestConfigurations; -import org.apache.hudi.operator.utils.TestData; -import org.apache.hudi.source.format.mor.MergeOnReadInputFormat; +import org.apache.hudi.table.format.mor.MergeOnReadInputFormat; import org.apache.hudi.util.StreamerUtil; +import org.apache.hudi.utils.TestConfigurations; +import org.apache.hudi.utils.TestData; import org.apache.flink.api.common.io.FileInputFormat; import org.apache.flink.api.common.io.InputFormat; diff --git a/hudi-flink/src/test/java/org/apache/hudi/source/format/TestInputFormat.java b/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java similarity index 96% rename from hudi-flink/src/test/java/org/apache/hudi/source/format/TestInputFormat.java rename to hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java index 1d92dbb68..c508a7fb5 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/source/format/TestInputFormat.java +++ b/hudi-flink/src/test/java/org/apache/hudi/table/format/TestInputFormat.java @@ -16,14 +16,14 @@ * limitations under the License. */ -package org.apache.hudi.source.format; +package org.apache.hudi.table.format; import org.apache.hudi.common.model.HoodieTableType; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.operator.utils.TestConfigurations; -import org.apache.hudi.operator.utils.TestData; -import org.apache.hudi.source.HoodieTableSource; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.table.HoodieTableSource; import org.apache.hudi.util.StreamerUtil; +import org.apache.hudi.utils.TestConfigurations; +import org.apache.hudi.utils.TestData; import org.apache.flink.api.common.io.InputFormat; import org.apache.flink.configuration.Configuration; diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/TestConfigurations.java b/hudi-flink/src/test/java/org/apache/hudi/utils/TestConfigurations.java similarity index 98% rename from hudi-flink/src/test/java/org/apache/hudi/operator/utils/TestConfigurations.java rename to hudi-flink/src/test/java/org/apache/hudi/utils/TestConfigurations.java index 358c88f62..e32b9c021 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/TestConfigurations.java +++ b/hudi-flink/src/test/java/org/apache/hudi/utils/TestConfigurations.java @@ -16,9 +16,9 @@ * limitations under the License. */ -package org.apache.hudi.operator.utils; +package org.apache.hudi.utils; -import org.apache.hudi.operator.FlinkOptions; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.streamer.FlinkStreamerConfig; import org.apache.hudi.utils.factory.CollectSinkTableFactory; import org.apache.hudi.utils.factory.ContinuousFileSourceFactory; diff --git a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/TestData.java b/hudi-flink/src/test/java/org/apache/hudi/utils/TestData.java similarity index 99% rename from hudi-flink/src/test/java/org/apache/hudi/operator/utils/TestData.java rename to hudi-flink/src/test/java/org/apache/hudi/utils/TestData.java index 3b63febdc..ab7f544bf 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/operator/utils/TestData.java +++ b/hudi-flink/src/test/java/org/apache/hudi/utils/TestData.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.hudi.operator.utils; +package org.apache.hudi.utils; import org.apache.hudi.client.FlinkTaskContextSupplier; import org.apache.hudi.client.common.HoodieFlinkEngineContext; @@ -25,7 +25,8 @@ import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.common.table.log.HoodieMergedLogRecordScanner; import org.apache.hudi.common.testutils.HoodieTestUtils; import org.apache.hudi.config.HoodieWriteConfig; -import org.apache.hudi.operator.FlinkOptions; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.sink.utils.StreamWriteFunctionWrapper; import org.apache.hudi.table.HoodieFlinkTable; import org.apache.avro.Schema; diff --git a/hudi-flink/src/test/java/org/apache/hudi/utils/TestUtils.java b/hudi-flink/src/test/java/org/apache/hudi/utils/TestUtils.java index fa021f3da..ecc86e720 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/utils/TestUtils.java +++ b/hudi-flink/src/test/java/org/apache/hudi/utils/TestUtils.java @@ -19,9 +19,9 @@ package org.apache.hudi.utils; import org.apache.hudi.common.table.HoodieTableMetaClient; -import org.apache.hudi.operator.FlinkOptions; -import org.apache.hudi.operator.StreamReadMonitoringFunction; -import org.apache.hudi.source.format.mor.MergeOnReadInputSplit; +import org.apache.hudi.configuration.FlinkOptions; +import org.apache.hudi.source.StreamReadMonitoringFunction; +import org.apache.hudi.table.format.mor.MergeOnReadInputSplit; import org.apache.hudi.util.StreamerUtil; import org.apache.flink.configuration.Configuration; diff --git a/hudi-flink/src/test/java/org/apache/hudi/utils/factory/CollectSinkTableFactory.java b/hudi-flink/src/test/java/org/apache/hudi/utils/factory/CollectSinkTableFactory.java index 902987ef7..d27679bad 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/utils/factory/CollectSinkTableFactory.java +++ b/hudi-flink/src/test/java/org/apache/hudi/utils/factory/CollectSinkTableFactory.java @@ -18,7 +18,7 @@ package org.apache.hudi.utils.factory; -import org.apache.hudi.operator.utils.TestConfigurations; +import org.apache.hudi.utils.TestConfigurations; import org.apache.flink.api.common.state.ListState; import org.apache.flink.api.common.state.ListStateDescriptor; diff --git a/hudi-flink/src/test/java/org/apache/hudi/utils/factory/ContinuousFileSourceFactory.java b/hudi-flink/src/test/java/org/apache/hudi/utils/factory/ContinuousFileSourceFactory.java index 171f82dc0..21128bedf 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/utils/factory/ContinuousFileSourceFactory.java +++ b/hudi-flink/src/test/java/org/apache/hudi/utils/factory/ContinuousFileSourceFactory.java @@ -18,7 +18,7 @@ package org.apache.hudi.utils.factory; -import org.apache.hudi.operator.FlinkOptions; +import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.utils.source.ContinuousFileSource; import org.apache.flink.configuration.Configuration; diff --git a/hudi-flink/src/test/java/org/apache/hudi/utils/source/ContinuousFileSource.java b/hudi-flink/src/test/java/org/apache/hudi/utils/source/ContinuousFileSource.java index 56f535b70..2f759b870 100644 --- a/hudi-flink/src/test/java/org/apache/hudi/utils/source/ContinuousFileSource.java +++ b/hudi-flink/src/test/java/org/apache/hudi/utils/source/ContinuousFileSource.java @@ -153,7 +153,7 @@ public class ContinuousFileSource implements StreamTableSource { private void loadDataBuffer() { this.dataBuffer = new ArrayList<>(); try (BufferedReader reader = - new BufferedReader(new FileReader(this.path.getPath()))) { + new BufferedReader(new FileReader(this.path.toString()))) { String line = reader.readLine(); while (line != null) { this.dataBuffer.add(line); diff --git a/hudi-flink/src/test/resources/META-INF/services/org.apache.flink.table.factories.TableFactory b/hudi-flink/src/test/resources/META-INF/services/org.apache.flink.table.factories.TableFactory index 994175038..87c3d990a 100644 --- a/hudi-flink/src/test/resources/META-INF/services/org.apache.flink.table.factories.TableFactory +++ b/hudi-flink/src/test/resources/META-INF/services/org.apache.flink.table.factories.TableFactory @@ -14,5 +14,5 @@ # See the License for the specific language governing permissions and # limitations under the License. -org.apache.hudi.factory.HoodieTableFactory +org.apache.hudi.table.HoodieTableFactory org.apache.hudi.utils.factory.ContinuousFileSourceFactory diff --git a/packaging/hudi-flink-bundle/pom.xml b/packaging/hudi-flink-bundle/pom.xml index b8a8cd1ef..255d0c0bf 100644 --- a/packaging/hudi-flink-bundle/pom.xml +++ b/packaging/hudi-flink-bundle/pom.xml @@ -31,6 +31,8 @@ true ${project.parent.basedir} + org.apache.hudi. + 3.1.0 @@ -70,6 +72,7 @@ org.apache.hudi:hudi-client-common org.apache.hudi:hudi-flink-client org.apache.hudi:hudi-flink_${scala.binary.version} + org.apache.hudi:hudi-hadoop-mr org.apache.hudi:hudi-timeline-service com.yammer.metrics:metrics-core @@ -81,7 +84,14 @@ org.apache.httpcomponents:httpcore org.apache.httpcomponents:fluent-hc org.antlr:stringtemplate + org.apache.parquet:parquet-avro + org.apache.parquet:parquet-hadoop + org.apache.parquet:parquet-column + org.apache.parquet:parquet-common + org.apache.parquet:parquet-format + org.apache.parquet:parquet-encoding + org.apache.avro:avro joda-time:joda-time com.fasterxml.jackson.core:jackson-annotations @@ -103,11 +113,25 @@ io.prometheus:simpleclient_pushgateway io.prometheus:simpleclient_common com.yammer.metrics:metrics-core - org.apache.flink:flink-connector-kafka_${scala.binary.version} - org.apache.flink:flink-connector-kafka-base_${scala.binary.version} org.apache.kafka:kafka_${scala.binary.version} com.101tec:zkclient - org.apache.kafka:kafka-clients + + + org.eclipse.jetty:* + org.eclipse.jetty.websocket:* + javax.servlet:javax.servlet-api + + org.apache.flink:flink-connector-kafka_${scala.binary.version} + org.apache.flink:flink-connector-kafka-base_${scala.binary.version} + org.apache.flink:flink-hadoop-compatibility_${scala.binary.version} + org.apache.flink:flink-avro + org.apache.flink:flink-json + org.apache.flink:flink-parquet_${scala.binary.version} + + org.apache.hive:hive-common + org.apache.hive:hive-exec + org.apache.hive:hive-metastore + org.apache.hive:hive-jdbc org.apache.hbase:hbase-client org.apache.hbase:hbase-common @@ -125,6 +149,34 @@ com.beust.jcommander. org.apache.hudi.com.beust.jcommander. + + org.apache.hive.jdbc. + ${flink.bundle.hive.shade.prefix}org.apache.hive.jdbc. + + + org.apache.hadoop.hive.metastore. + ${flink.bundle.hive.shade.prefix}org.apache.hadoop.hive.metastore. + + + org.apache.hive.common. + ${flink.bundle.hive.shade.prefix}org.apache.hive.common. + + + org.apache.hadoop.hive.common. + ${flink.bundle.hive.shade.prefix}org.apache.hadoop.hive.common. + + + org.apache.hadoop.hive.conf. + ${flink.bundle.hive.shade.prefix}org.apache.hadoop.hive.conf. + + + org.apache.hive.service. + ${flink.bundle.hive.shade.prefix}org.apache.hive.service. + + + org.apache.hadoop.hive.service. + ${flink.bundle.hive.shade.prefix}org.apache.hadoop.hive.service. + com.codahale.metrics. org.apache.hudi.com.codahale.metrics. @@ -133,6 +185,10 @@ org.apache.commons.codec. org.apache.hudi.org.apache.commons.codec. + + org.eclipse.jetty. + org.apache.hudi.org.apache.jetty. + @@ -142,6 +198,8 @@ META-INF/*.DSA META-INF/*.RSA META-INF/services/javax.* + + META-INF/services/org.apache.flink.table.factories.Factory @@ -183,6 +241,21 @@ hudi-flink_${scala.binary.version} ${project.version} + + org.apache.hudi + hudi-hadoop-mr + ${project.version} + + + org.apache.hudi + hudi-timeline-service + ${project.version} + + + javax.servlet + javax.servlet-api + ${javax.servlet.version} + @@ -196,6 +269,30 @@ ${flink.version} compile + + org.apache.flink + flink-hadoop-compatibility_${scala.binary.version} + ${flink.version} + compile + + + org.apache.flink + flink-avro + ${flink.version} + compile + + + org.apache.flink + flink-parquet_${scala.binary.version} + ${flink.version} + compile + + + org.apache.flink + flink-json + ${flink.version} + compile + @@ -208,9 +305,69 @@ org.apache.avro avro + + 1.10.0 compile + + + ${hive.groupid} + hive-exec + ${hive.version} + + + javax.mail + mail + + + org.eclipse.jetty.aggregate + * + + + + + ${hive.groupid} + hive-metastore + ${hive.version} + + + javax.servlet + * + + + javax.servlet.jsp + * + + + + + ${hive.groupid} + hive-jdbc + ${hive.version} + + + javax.servlet + * + + + javax.servlet.jsp + * + + + + + ${hive.groupid} + hive-common + ${hive.version} + + + org.eclipse.jetty.orbit + javax.servlet + + + + joda-time joda-time