hudi-agent commented on code in PR #19805:
URL: https://github.com/apache/hudi/pull/19805#discussion_r3909223718
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/configuration/TestOptionsResolver.java:
##########
@@ -407,6 +407,23 @@ void testConflictResolutionStrategies() {
OptionsResolver.getConflictResolutionStrategy(conf));
}
+ @Test
+ void testPartitionedRLIWithRocksDBBackend() {
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.RECORD_LEVEL_INDEX.name());
+ conf.set(FlinkOptions.INDEX_RLI_BACKEND_TYPE, "mdt");
+ assertFalse(OptionsResolver.isTimeBoundedRLIBootstrapEnabled(conf));
+
+ conf.set(FlinkOptions.INDEX_RLI_BACKEND_TYPE, "rocksdb");
+ assertTrue(OptionsResolver.isTimeBoundedRLIBootstrapEnabled(conf));
Review Comment:
🤖 With `index.rli.cache.rocksdb.bootstrap.days` now defaulting to `-1`,
`isTimeBoundedRLIBootstrapEnabled` requires `days > 0`, but this test never
sets it — so `conf.get(...)` returns `-1` and this `assertTrue` would actually
fail. Could you `conf.set(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS,
<positive>)` before asserting enabled?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/TimeBoundedRLIBootstrapOperator.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.bootstrap;
+
+import org.apache.hudi.client.common.HoodieFlinkEngineContext;
+import org.apache.hudi.client.model.HoodieFlinkInternalRow;
+import org.apache.hudi.common.data.HoodiePairData;
+import org.apache.hudi.common.function.SerializableFunctionUnchecked;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Functions;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.common.util.hash.BucketIndexUtil;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.metadata.HoodieBackedTableMetadata;
+import org.apache.hudi.metadata.MetadataPartitionType;
+import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.RuntimeContextUtils;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.time.LocalDate;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Bootstrap operator that preload of time bounded partitioned record level
index (RLI) data
+ * from the metadata table.
+ *
+ * <p>Only data table partitions that fall within the last {@link
FlinkOptions#INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS}
+ * days are eagerly preloaded; the partition path of each partition is parsed
as a date using
+ * {@link FlinkOptions#PARTITION_FORMAT} (default {@link
FlinkOptions#PARTITION_FORMAT_DAY}) to
+ * determine whether it falls inside the window. Partitions outside the
window, and partitions whose
+ * path cannot be parsed as a date, are skipped here and are expected to be
loaded on demand later.
+ *
+ * <p>Setting {@link FlinkOptions#INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS} to
{@code 0} disables preloading
+ * entirely, which is the expected fallback for non-temporal (non
date-partitioned) tables.
+ */
+@Slf4j
+public class TimeBoundedRLIBootstrapOperator
+ extends AbstractBootstrapOperator {
+
+ private transient HoodieBackedTableMetadata tableMetadata;
+ private transient long loadedCnt;
+ private int parallelism;
+ private int taskID;
+ /**
+ * Functions for calculating the task partition to dispatch.
Review Comment:
🤖 nit: the javadoc says setting this to `0` disables preloading, but the
default is now `-1` and the check is `<= 0`. Could you reword to say any
non-positive value disables it to match the code?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/TimeBoundedRLIBootstrapOperator.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.sink.bootstrap;
+
+import org.apache.hudi.client.common.HoodieFlinkEngineContext;
+import org.apache.hudi.client.model.HoodieFlinkInternalRow;
+import org.apache.hudi.common.data.HoodiePairData;
+import org.apache.hudi.common.function.SerializableFunctionUnchecked;
+import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieRecordGlobalLocation;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Functions;
+import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.common.util.hash.BucketIndexUtil;
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.metadata.HoodieBackedTableMetadata;
+import org.apache.hudi.metadata.MetadataPartitionType;
+import org.apache.hudi.util.StreamerUtil;
+import org.apache.hudi.utils.RuntimeContextUtils;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import java.time.LocalDate;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Bootstrap operator that preload of time bounded partitioned record level
index (RLI) data
+ * from the metadata table.
+ *
+ * <p>Only data table partitions that fall within the last {@link
FlinkOptions#INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS}
+ * days are eagerly preloaded; the partition path of each partition is parsed
as a date using
+ * {@link FlinkOptions#PARTITION_FORMAT} (default {@link
FlinkOptions#PARTITION_FORMAT_DAY}) to
+ * determine whether it falls inside the window. Partitions outside the
window, and partitions whose
+ * path cannot be parsed as a date, are skipped here and are expected to be
loaded on demand later.
+ *
+ * <p>Setting {@link FlinkOptions#INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS} to
{@code 0} disables preloading
+ * entirely, which is the expected fallback for non-temporal (non
date-partitioned) tables.
+ */
+@Slf4j
+public class TimeBoundedRLIBootstrapOperator
+ extends AbstractBootstrapOperator {
+
+ private transient HoodieBackedTableMetadata tableMetadata;
+ private transient long loadedCnt;
+ private int parallelism;
+ private int taskID;
+ /**
+ * Functions for calculating the task partition to dispatch.
+ */
+ @VisibleForTesting
+ Functions.Function3<Integer, String, Integer, Integer> partitionIndexFunc;
+
+ public TimeBoundedRLIBootstrapOperator(Configuration conf) {
+ super(conf);
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws
Exception {
+ loadedCnt = 0;
+ this.taskID =
RuntimeContextUtils.getIndexOfThisSubtask(getRuntimeContext());
+
+ int bootstrapDays =
conf.get(FlinkOptions.INDEX_RLI_CACHE_ROCKSDB_BOOTSTRAP_DAYS);
+ if (bootstrapDays <= 0) {
+ log.info("Skip preloading partitioned RLI records because bootstrap days
is configured as {}, taskId = {}",
+ bootstrapDays, taskID);
+ waitForBootstrapReady(taskID);
+ return;
+ }
+
+ HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(conf);
+ this.tableMetadata = createTableMetadata(metaClient);
+
+ this.parallelism =
RuntimeContextUtils.getNumberOfParallelSubtasks(getRuntimeContext());
+ this.partitionIndexFunc =
BucketIndexUtil.getPartitionIndexFunc(parallelism);
+ preLoadPartitionedRLIRecords(metaClient.getTableConfig(), bootstrapDays);
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeMetadataTable();
+ super.close();
+ }
+
+ // -------------------------------------------------------------------------
+ // Utilities
+ // -------------------------------------------------------------------------
+
+ @VisibleForTesting
+ HoodieBackedTableMetadata createTableMetadata(HoodieTableMetaClient
metaClient) {
+ return new HoodieBackedTableMetadata(
+ HoodieFlinkEngineContext.DEFAULT,
+ metaClient.getStorage(),
+ StreamerUtil.metadataConfig(conf),
+ conf.get(FlinkOptions.PATH));
+ }
+
+ private void preLoadPartitionedRLIRecords(HoodieTableConfig tableConfig, int
bootstrapDays) {
+ if (!tableMetadata.enabled()) {
+ if (tableConfig.isMetadataTableAvailable()) {
+ throw new RuntimeException("Can not initialize the table metadata");
+ }
+ log.info("Skip preloading partitioned RLI records because table metadata
is not initialized, taskId = {}", taskID);
+ waitForBootstrapReady(taskID);
+ closeMetadataTable();
+ return;
+ }
+
+ if
(!tableConfig.isMetadataPartitionAvailable(MetadataPartitionType.RECORD_INDEX))
{
+ log.info("Skip preloading partitioned RLI records because record index
is not available yet, taskId = {}", taskID);
+ waitForBootstrapReady(taskID);
+ closeMetadataTable();
+ return;
+ }
+
+ Map<String, List<FileSlice>> partitionedFileGroups =
+
tableMetadata.getBucketizedFileGroupsForPartitionedRLI(MetadataPartitionType.RECORD_INDEX);
+ List<String> partitionsInWindow =
filterPartitionsInWindow(partitionedFileGroups.keySet(), bootstrapDays);
+
+ log.info("Start preloading partitioned RLI records from metadata table for
{}/{} partitions within the last {} days, "
+ + "taskId = {}, parallelism = {}",
+ partitionsInWindow.size(), partitionedFileGroups.size(),
bootstrapDays, taskID, parallelism);
+
+ long startTime = System.currentTimeMillis();
+ for (String partitionPath : partitionsInWindow) {
+ preLoadPartition(partitionPath,
partitionedFileGroups.get(partitionPath), taskID, parallelism);
+ }
+ long costMs = System.currentTimeMillis() - startTime;
+ log.info("Finish preloading partitioned RLI records, total records: {},
cost: {} ms, taskId = {}", loadedCnt, costMs, taskID);
Review Comment:
🤖 nit: preLoadPartition takes taskID and parallelism as parameters, but both
are already instance fields set in initializeState. Could you drop the params
and read the fields directly to avoid the confusion of two sources of truth?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestPipelines.java:
##########
@@ -101,6 +105,27 @@ void testBootstrapPipelineSelection() {
assertEquals(3, streaming.getParallelism());
}
+ @Test
+ void
testPartitionedRLIWithRocksDBBackendUsesPartitionedRLIBootstrapOperator() {
+ Configuration conf = defaultConf();
+ conf.set(FlinkOptions.INDEX_GLOBAL_ENABLED, false);
+ conf.set(FlinkOptions.INDEX_TYPE,
HoodieIndex.IndexType.RECORD_LEVEL_INDEX.name());
+ conf.set(FlinkOptions.INDEX_RLI_BACKEND_TYPE, "rocksdb");
+ conf.set(FlinkOptions.INDEX_BOOTSTRAP_ENABLED, true);
+ DataStream<RowData> input = rowDataInput();
+
+ DataStream<HoodieFlinkInternalRow> streaming =
+ Pipelines.bootstrap(conf, TestConfigurations.ROW_TYPE, input, false,
false);
+
+ assertEquals("index_bootstrap", streaming.getTransformation().getName());
+ assertInstanceOf(TimeBoundedRLIBootstrapOperator.class,
bootstrapOperator(streaming));
Review Comment:
🤖 Same root cause as the OptionsResolver test: `defaultConf()` doesn't set
`index.rli.cache.rocksdb.bootstrap.days`, so it defaults to `-1`,
`isTimeBoundedRLIBootstrapEnabled` returns false, and `streamBootstrap` selects
`BootstrapOperator` instead — so this
`assertInstanceOf(TimeBoundedRLIBootstrapOperator.class, ...)` would fail.
Setting a positive bootstrap-days on `conf` should fix it.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
