You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/05/06 04:55:25 UTC
[GitHub] [flink] JingsongLi opened a new pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
JingsongLi opened a new pull request #12004:
URL: https://github.com/apache/flink/pull/12004
## What is the purpose of the change
Implement a hive streaming source, it monitor partitions of hive meta store. Streaming reading.
## Brief change log
- Refactor ContinuousFileReaderOperator to read generic split to reuse this operator.
- HiveTableInputFormat implements CheckpointableInputFormat for streaming reading
- Support streaming read for hive partitioned source
## Verifying this change
Manually testing.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`:no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented?JavaDocs
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624441821
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299 (Wed May 06 04:58:13 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422753767
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.flink.table.filesystem;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import java.time.Duration;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds configuration constants used by filesystem(Including hive) connector.
+ */
+public class FileSystemOptions {
+
+ public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
+ key("streaming-source.enable")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Enable streaming source or not.");
+
+ public static final ConfigOption<Duration> STREAMING_SOURCE_MONITOR_INTERVAL =
+ key("streaming-source.monitor-interval")
+ .durationType()
+ .defaultValue(Duration.ofMillis(1))
+ .withDescription("The minimum interval allowed between consecutive partition/file discovery.");
+
+ public static final ConfigOption<String> STREAMING_SOURCE_CONSUME_ORDER =
+ key("streaming-source.consume-order")
+ .stringType()
+ .defaultValue("create-time")
+ .withDescription("The consume order of streaming source," +
+ " support create-time and partition-time." +
+ " create-time compare partition/file creation time;" +
+ " partition-time compare time represented by partition name.");
+
+ public static final ConfigOption<String> STREAMING_SOURCE_CONSUME_START_OFFSET =
+ key("streaming-source.consume-start-offset")
+ .stringType()
+ .defaultValue("1970-00-00")
+ .withDescription("Start offset for streaming consuming." +
+ " How to parse and compare offsets depends on your order." +
+ " For create-time and partition-time, should be a timestamp string.");
+
+ public static final ConfigOption<String> PARTITION_TIME_EXTRACTOR_TYPE =
+ key("partition.time-extractor.type")
+ .stringType()
+ .defaultValue("timestamp")
+ .withDescription("Time extractor to extract time from partition values." +
+ " Support timestamp and custom." +
Review comment:
+1
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422751974
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.flink.table.filesystem;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import java.time.Duration;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds configuration constants used by filesystem(Including hive) connector.
+ */
+public class FileSystemOptions {
+
+ public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
Review comment:
No, but streaming execution mode can run both streaming source and batch source.
Batch execution mode is a subset of streaming execution.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422462431
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,280 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.connectors.hive.read.PartitionStrategy.PartitionStrategyFactory;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * This is the single (non-parallel) monitoring task which takes a {@link HiveTableInputFormat},
+ * it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring partitions of hive meta store.</li>
+ * <li>Deciding which partitions should be further read and processed.</li>
+ * <li>Creating the {@link HiveTableInputSplit splits} corresponding to those partitions.</li>
+ * <li>Assigning them to downstream tasks for further processing.</li>
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ContinuousFileReaderOperator}
+ * which can have parallelism greater than one.
+ *
+ * <p><b>IMPORTANT NOTE: </b> Splits are forwarded downstream for reading in ascending partition time order,
+ * based on the partition time of the partitions they belong to.
+ */
+public class HiveContinuousMonitoringFunction
+ extends RichSourceFunction<TimestampedHiveInputSplit>
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveContinuousMonitoringFunction.class);
+
+ /** The parallelism of the downstream readers. */
+ private final int readerParallelism;
+
+ /** The interval between consecutive path scans. */
+ private final long interval;
+
+ private final HiveShim hiveShim;
+
+ private final JobConfWrapper conf;
+
+ private final ObjectPath tablePath;
+
+ private final List<String> partitionKeys;
+
+ private final String[] fieldNames;
+
+ private final DataType[] fieldTypes;
+
+ private final long startupTimestampMillis;
+
+ private final PartitionStrategyFactory partStrategyFactory;
+
+ private volatile boolean isRunning = true;
+
+ /** The maximum partition read time seen so far. */
+ private volatile long currentReadTime;
+
+ private transient PartitionStrategy strategy;
+
+ private transient Object checkpointLock;
+
+ private transient ListState<Long> currReadTimeState;
+
+ private transient ListState<List<List<String>>> distinctPartsState;
+
+ private transient IMetaStoreClient client;
+
+ private transient Properties tableProps;
+
+ private transient String defaultPartitionName;
+
+ private transient Set<List<String>> distinctPartitions;
+
+ public HiveContinuousMonitoringFunction(
+ HiveShim hiveShim,
+ JobConf conf,
+ ObjectPath tablePath,
+ CatalogTable catalogTable,
+ long startupTimestampMillis,
+ PartitionStrategyFactory partStrategyFactory,
+ int readerParallelism,
+ long interval) {
+ this.hiveShim = hiveShim;
+ this.conf = new JobConfWrapper(conf);
+ this.tablePath = tablePath;
+ this.partitionKeys = catalogTable.getPartitionKeys();
+ this.fieldNames = catalogTable.getSchema().getFieldNames();
+ this.fieldTypes = catalogTable.getSchema().getFieldDataTypes();
+ this.startupTimestampMillis = startupTimestampMillis;
+ this.partStrategyFactory = partStrategyFactory;
+
+ this.interval = interval;
+ this.readerParallelism = Math.max(readerParallelism, 1);
+ this.currentReadTime = Long.MIN_VALUE;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.strategy = partStrategyFactory.createStrategy(getRuntimeContext().getUserCodeClassLoader());
+
+ this.currReadTimeState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ LongSerializer.INSTANCE
+ )
+ );
+ this.distinctPartsState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
+ )
+ );
+
+ this.client = this.hiveShim.getHiveMetastoreClient(new HiveConf(conf.conf(), HiveConf.class));
+
+ Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
+ this.tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, hiveTable);
+ this.defaultPartitionName = conf.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+ HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
+
+ this.distinctPartitions = new HashSet<>();
+ if (context.isRestored()) {
+ LOG.info("Restoring state for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.currReadTimeState.get().iterator().next();
+ this.distinctPartitions.addAll(this.distinctPartsState.get().iterator().next());
+ } else {
+ LOG.info("No state to restore for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.startupTimestampMillis;
+ }
+ }
+
+ @Override
+ public void run(SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ checkpointLock = context.getCheckpointLock();
+ while (isRunning) {
+ synchronized (checkpointLock) {
+ monitorAndForwardSplits(context);
+ }
+ Thread.sleep(interval);
+ }
+ }
+
+ private void monitorAndForwardSplits(
+ SourceContext<TimestampedHiveInputSplit> context) throws IOException, TException {
+ assert (Thread.holdsLock(checkpointLock));
+
+ List<Partition> partitions = client.listPartitionsByFilter(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ strategy.generateFetchFilter(partitionKeys, currentReadTime),
+ (short) -1);
+
+ if (partitions.isEmpty()) {
+ return;
+ }
+
+ long maxTime = Long.MIN_VALUE;
+ for (Partition partition : partitions) {
Review comment:
Good point, yes, will sort.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lirui-apache commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
lirui-apache commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422756417
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/TimestampedHiveInputSplit.java
##########
@@ -0,0 +1,232 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.streaming.api.functions.source.TimestampedInputSplit;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.lang.reflect.Type;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * A {@link HiveTableInputSplit} with {@link TimestampedInputSplit}.
+ * Kryo serializer can not deal with hadoop split, need specific type information factory.
+ */
+@TypeInfo(TimestampedHiveInputSplit.SplitTypeInfoFactory.class)
+public class TimestampedHiveInputSplit extends HiveTableInputSplit implements TimestampedInputSplit {
+
+ private static final long serialVersionUID = 1L;
+
+ /** The modification time of the file this split belongs to. */
+ private final long modificationTime;
+
+ /**
+ * The state of the split. This information is used when
+ * restoring from a checkpoint and allows to resume reading the
+ * underlying file from the point we left off.
+ * */
+ private Serializable splitState;
+
+ public TimestampedHiveInputSplit(
+ long modificationTime,
+ HiveTableInputSplit split) {
+ super(
+ split.getSplitNumber(),
+ split.getHadoopInputSplit(),
+ split.getJobConf(),
+ split.getHiveTablePartition());
+ this.modificationTime = modificationTime;
+ }
+
+ @Override
+ public void setSplitState(Serializable state) {
+ this.splitState = state;
+ }
+
+ @Override
+ public Serializable getSplitState() {
+ return this.splitState;
+ }
+
+ @Override
+ public long getModificationTime() {
+ return modificationTime;
+ }
+
+ @Override
+ public int compareTo(TimestampedInputSplit o) {
+ TimestampedHiveInputSplit split = (TimestampedHiveInputSplit) o;
+ int modTimeComp = Long.compare(this.modificationTime, split.modificationTime);
+ if (modTimeComp != 0L) {
+ return modTimeComp;
+ }
+
+ int pathComp = this.hiveTablePartition.getStorageDescriptor().compareTo(
+ split.hiveTablePartition.getStorageDescriptor());
+
+ return pathComp != 0 ? pathComp :
+ this.getSplitNumber() - o.getSplitNumber();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
Review comment:
I think the best practice is to avoid such implementations. And if for some reason it has to be done this way, we should clarify it in the java doc, according to the suggestions of `Comparable`:
```
* <p>It is strongly recommended, but <i>not</i> strictly required that
* <tt>(x.compareTo(y)==0) == (x.equals(y))</tt>. Generally speaking, any
* class that implements the <tt>Comparable</tt> interface and violates
* this condition should clearly indicate this fact. The recommended
* language is "Note: this class has a natural ordering that is
* inconsistent with equals."
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
}, {
"hash" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"triggerType" : "PUSH"
}, {
"hash" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929",
"triggerID" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"triggerType" : "PUSH"
}, {
"hash" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937",
"triggerID" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"triggerType" : "PUSH"
}, {
"hash" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=946",
"triggerID" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"triggerType" : "PUSH"
}, {
"hash" : "bffea7d6cacc503cc630c8b3ddf77f4093554e84",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=967",
"triggerID" : "bffea7d6cacc503cc630c8b3ddf77f4093554e84",
"triggerType" : "PUSH"
}, {
"hash" : "48017434485d6059dafddc8617bf7f75fa7b3892",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "48017434485d6059dafddc8617bf7f75fa7b3892",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3ac79a41395792dfecefc87a19c58468e91ede8d UNKNOWN
* 006610cb59c3665b510485c6d6ad883d91fc3e6d Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=946)
* bffea7d6cacc503cc630c8b3ddf77f4093554e84 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=967)
* 48017434485d6059dafddc8617bf7f75fa7b3892 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 42953325bb1d6a447ef53e020d3cc311a8c986c6 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lirui-apache commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
lirui-apache commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r420742231
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveOptions.java
##########
@@ -45,4 +47,34 @@
key("table.exec.hive.infer-source-parallelism.max")
.defaultValue(1000)
.withDescription("Sets max infer parallelism for source operator.");
+
+ public static final ConfigOption<Boolean> HIVE_STREAMING_SOURCE_ENABLE =
+ key("hive.streaming-source.enable")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("");
Review comment:
I think these properties are user-facing right? Need to add descriptions for them.
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,280 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.connectors.hive.read.PartitionStrategy.PartitionStrategyFactory;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * This is the single (non-parallel) monitoring task which takes a {@link HiveTableInputFormat},
+ * it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring partitions of hive meta store.</li>
+ * <li>Deciding which partitions should be further read and processed.</li>
+ * <li>Creating the {@link HiveTableInputSplit splits} corresponding to those partitions.</li>
+ * <li>Assigning them to downstream tasks for further processing.</li>
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ContinuousFileReaderOperator}
+ * which can have parallelism greater than one.
+ *
+ * <p><b>IMPORTANT NOTE: </b> Splits are forwarded downstream for reading in ascending partition time order,
+ * based on the partition time of the partitions they belong to.
+ */
+public class HiveContinuousMonitoringFunction
+ extends RichSourceFunction<TimestampedHiveInputSplit>
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveContinuousMonitoringFunction.class);
+
+ /** The parallelism of the downstream readers. */
+ private final int readerParallelism;
+
+ /** The interval between consecutive path scans. */
+ private final long interval;
+
+ private final HiveShim hiveShim;
+
+ private final JobConfWrapper conf;
+
+ private final ObjectPath tablePath;
+
+ private final List<String> partitionKeys;
+
+ private final String[] fieldNames;
+
+ private final DataType[] fieldTypes;
+
+ private final long startupTimestampMillis;
+
+ private final PartitionStrategyFactory partStrategyFactory;
+
+ private volatile boolean isRunning = true;
+
+ /** The maximum partition read time seen so far. */
+ private volatile long currentReadTime;
+
+ private transient PartitionStrategy strategy;
+
+ private transient Object checkpointLock;
+
+ private transient ListState<Long> currReadTimeState;
+
+ private transient ListState<List<List<String>>> distinctPartsState;
+
+ private transient IMetaStoreClient client;
+
+ private transient Properties tableProps;
+
+ private transient String defaultPartitionName;
+
+ private transient Set<List<String>> distinctPartitions;
+
+ public HiveContinuousMonitoringFunction(
+ HiveShim hiveShim,
+ JobConf conf,
+ ObjectPath tablePath,
+ CatalogTable catalogTable,
+ long startupTimestampMillis,
+ PartitionStrategyFactory partStrategyFactory,
+ int readerParallelism,
+ long interval) {
+ this.hiveShim = hiveShim;
+ this.conf = new JobConfWrapper(conf);
+ this.tablePath = tablePath;
+ this.partitionKeys = catalogTable.getPartitionKeys();
+ this.fieldNames = catalogTable.getSchema().getFieldNames();
+ this.fieldTypes = catalogTable.getSchema().getFieldDataTypes();
+ this.startupTimestampMillis = startupTimestampMillis;
+ this.partStrategyFactory = partStrategyFactory;
+
+ this.interval = interval;
+ this.readerParallelism = Math.max(readerParallelism, 1);
+ this.currentReadTime = Long.MIN_VALUE;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.strategy = partStrategyFactory.createStrategy(getRuntimeContext().getUserCodeClassLoader());
+
+ this.currReadTimeState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ LongSerializer.INSTANCE
+ )
+ );
+ this.distinctPartsState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
+ )
+ );
+
+ this.client = this.hiveShim.getHiveMetastoreClient(new HiveConf(conf.conf(), HiveConf.class));
+
+ Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
+ this.tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, hiveTable);
+ this.defaultPartitionName = conf.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+ HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
+
+ this.distinctPartitions = new HashSet<>();
+ if (context.isRestored()) {
+ LOG.info("Restoring state for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.currReadTimeState.get().iterator().next();
+ this.distinctPartitions.addAll(this.distinctPartsState.get().iterator().next());
+ } else {
+ LOG.info("No state to restore for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.startupTimestampMillis;
+ }
+ }
+
+ @Override
+ public void run(SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ checkpointLock = context.getCheckpointLock();
+ while (isRunning) {
+ synchronized (checkpointLock) {
+ monitorAndForwardSplits(context);
+ }
+ Thread.sleep(interval);
+ }
+ }
+
+ private void monitorAndForwardSplits(
+ SourceContext<TimestampedHiveInputSplit> context) throws IOException, TException {
+ assert (Thread.holdsLock(checkpointLock));
+
+ List<Partition> partitions = client.listPartitionsByFilter(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ strategy.generateFetchFilter(partitionKeys, currentReadTime),
+ (short) -1);
+
+ if (partitions.isEmpty()) {
+ return;
+ }
+
+ long maxTime = Long.MIN_VALUE;
+ for (Partition partition : partitions) {
+ List<String> partSpec = partition.getValues();
+ if (!this.distinctPartitions.contains(partSpec)) {
+ this.distinctPartitions.add(partition.getValues());
+ long time = this.strategy.extractPartTime(partitionKeys, partition.getValues());
+ if (time > maxTime) {
+ maxTime = time;
+ }
+ HiveTableInputSplit[] splits = HiveTableInputFormat.createInputSplits(
+ this.readerParallelism,
+ Collections.singletonList(toHiveTablePartition(partition)),
+ this.conf.conf());
+ for (HiveTableInputSplit split : splits) {
+ context.collect(new TimestampedHiveInputSplit(time, split));
+ }
+ }
+ }
+ this.currentReadTime = maxTime;
+
+ this.distinctPartitions.removeIf(partSpec -> this.strategy.canExpireForDistinct(
+ this.strategy.extractPartTime(partitionKeys, partSpec),
+ this.currentReadTime));
+ }
+
+ private HiveTablePartition toHiveTablePartition(Partition p) {
+ return HiveTableSource.toHiveTablePartition(
+ partitionKeys, fieldNames, fieldTypes, hiveShim, tableProps, defaultPartitionName, p);
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ Preconditions.checkState(this.currReadTimeState != null,
+ "The " + getClass().getSimpleName() + " state has not been properly initialized.");
+
+ this.currReadTimeState.clear();
+ this.currReadTimeState.add(this.currentReadTime);
+
+ this.distinctPartsState.clear();
+ this.distinctPartsState.add(new ArrayList<>(this.distinctPartitions));
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("{} checkpointed {}.", getClass().getSimpleName(), currentReadTime);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
Review comment:
Need to close the HMS client somewhere.
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,280 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.connectors.hive.read.PartitionStrategy.PartitionStrategyFactory;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * This is the single (non-parallel) monitoring task which takes a {@link HiveTableInputFormat},
+ * it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring partitions of hive meta store.</li>
+ * <li>Deciding which partitions should be further read and processed.</li>
+ * <li>Creating the {@link HiveTableInputSplit splits} corresponding to those partitions.</li>
+ * <li>Assigning them to downstream tasks for further processing.</li>
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ContinuousFileReaderOperator}
+ * which can have parallelism greater than one.
+ *
+ * <p><b>IMPORTANT NOTE: </b> Splits are forwarded downstream for reading in ascending partition time order,
+ * based on the partition time of the partitions they belong to.
+ */
+public class HiveContinuousMonitoringFunction
+ extends RichSourceFunction<TimestampedHiveInputSplit>
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveContinuousMonitoringFunction.class);
+
+ /** The parallelism of the downstream readers. */
+ private final int readerParallelism;
+
+ /** The interval between consecutive path scans. */
+ private final long interval;
+
+ private final HiveShim hiveShim;
+
+ private final JobConfWrapper conf;
+
+ private final ObjectPath tablePath;
+
+ private final List<String> partitionKeys;
+
+ private final String[] fieldNames;
+
+ private final DataType[] fieldTypes;
+
+ private final long startupTimestampMillis;
+
+ private final PartitionStrategyFactory partStrategyFactory;
+
+ private volatile boolean isRunning = true;
+
+ /** The maximum partition read time seen so far. */
+ private volatile long currentReadTime;
+
+ private transient PartitionStrategy strategy;
+
+ private transient Object checkpointLock;
+
+ private transient ListState<Long> currReadTimeState;
+
+ private transient ListState<List<List<String>>> distinctPartsState;
+
+ private transient IMetaStoreClient client;
+
+ private transient Properties tableProps;
+
+ private transient String defaultPartitionName;
+
+ private transient Set<List<String>> distinctPartitions;
+
+ public HiveContinuousMonitoringFunction(
+ HiveShim hiveShim,
+ JobConf conf,
+ ObjectPath tablePath,
+ CatalogTable catalogTable,
+ long startupTimestampMillis,
+ PartitionStrategyFactory partStrategyFactory,
+ int readerParallelism,
+ long interval) {
+ this.hiveShim = hiveShim;
+ this.conf = new JobConfWrapper(conf);
+ this.tablePath = tablePath;
+ this.partitionKeys = catalogTable.getPartitionKeys();
+ this.fieldNames = catalogTable.getSchema().getFieldNames();
+ this.fieldTypes = catalogTable.getSchema().getFieldDataTypes();
+ this.startupTimestampMillis = startupTimestampMillis;
+ this.partStrategyFactory = partStrategyFactory;
+
+ this.interval = interval;
+ this.readerParallelism = Math.max(readerParallelism, 1);
+ this.currentReadTime = Long.MIN_VALUE;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.strategy = partStrategyFactory.createStrategy(getRuntimeContext().getUserCodeClassLoader());
+
+ this.currReadTimeState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ LongSerializer.INSTANCE
+ )
+ );
+ this.distinctPartsState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
+ )
+ );
+
+ this.client = this.hiveShim.getHiveMetastoreClient(new HiveConf(conf.conf(), HiveConf.class));
+
+ Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
+ this.tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, hiveTable);
+ this.defaultPartitionName = conf.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+ HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
+
+ this.distinctPartitions = new HashSet<>();
+ if (context.isRestored()) {
+ LOG.info("Restoring state for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.currReadTimeState.get().iterator().next();
+ this.distinctPartitions.addAll(this.distinctPartsState.get().iterator().next());
+ } else {
+ LOG.info("No state to restore for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.startupTimestampMillis;
+ }
+ }
+
+ @Override
+ public void run(SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ checkpointLock = context.getCheckpointLock();
+ while (isRunning) {
+ synchronized (checkpointLock) {
+ monitorAndForwardSplits(context);
+ }
+ Thread.sleep(interval);
+ }
+ }
+
+ private void monitorAndForwardSplits(
+ SourceContext<TimestampedHiveInputSplit> context) throws IOException, TException {
+ assert (Thread.holdsLock(checkpointLock));
+
+ List<Partition> partitions = client.listPartitionsByFilter(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ strategy.generateFetchFilter(partitionKeys, currentReadTime),
+ (short) -1);
+
+ if (partitions.isEmpty()) {
+ return;
+ }
+
+ long maxTime = Long.MIN_VALUE;
+ for (Partition partition : partitions) {
+ List<String> partSpec = partition.getValues();
+ if (!this.distinctPartitions.contains(partSpec)) {
+ this.distinctPartitions.add(partition.getValues());
+ long time = this.strategy.extractPartTime(partitionKeys, partition.getValues());
+ if (time > maxTime) {
+ maxTime = time;
+ }
+ HiveTableInputSplit[] splits = HiveTableInputFormat.createInputSplits(
+ this.readerParallelism,
+ Collections.singletonList(toHiveTablePartition(partition)),
+ this.conf.conf());
+ for (HiveTableInputSplit split : splits) {
+ context.collect(new TimestampedHiveInputSplit(time, split));
+ }
+ }
+ }
+ this.currentReadTime = maxTime;
+
+ this.distinctPartitions.removeIf(partSpec -> this.strategy.canExpireForDistinct(
+ this.strategy.extractPartTime(partitionKeys, partSpec),
+ this.currentReadTime));
+ }
+
+ private HiveTablePartition toHiveTablePartition(Partition p) {
+ return HiveTableSource.toHiveTablePartition(
+ partitionKeys, fieldNames, fieldTypes, hiveShim, tableProps, defaultPartitionName, p);
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ Preconditions.checkState(this.currReadTimeState != null,
+ "The " + getClass().getSimpleName() + " state has not been properly initialized.");
+
+ this.currReadTimeState.clear();
+ this.currReadTimeState.add(this.currentReadTime);
+
+ this.distinctPartsState.clear();
+ this.distinctPartsState.add(new ArrayList<>(this.distinctPartitions));
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("{} checkpointed {}.", getClass().getSimpleName(), currentReadTime);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+
+ if (checkpointLock != null) {
+ synchronized (checkpointLock) {
+ currentReadTime = Long.MAX_VALUE;
+ isRunning = false;
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ if (checkpointLock != null) {
+ // this is to cover the case where cancel() is called before the run()
Review comment:
The `checkpointLock ` is assigned in `run()`. So if cancel is called before run, the `checkpointLock ` should always be null?
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,280 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.connectors.hive.read.PartitionStrategy.PartitionStrategyFactory;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * This is the single (non-parallel) monitoring task which takes a {@link HiveTableInputFormat},
+ * it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring partitions of hive meta store.</li>
+ * <li>Deciding which partitions should be further read and processed.</li>
+ * <li>Creating the {@link HiveTableInputSplit splits} corresponding to those partitions.</li>
+ * <li>Assigning them to downstream tasks for further processing.</li>
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ContinuousFileReaderOperator}
+ * which can have parallelism greater than one.
+ *
+ * <p><b>IMPORTANT NOTE: </b> Splits are forwarded downstream for reading in ascending partition time order,
+ * based on the partition time of the partitions they belong to.
+ */
+public class HiveContinuousMonitoringFunction
+ extends RichSourceFunction<TimestampedHiveInputSplit>
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveContinuousMonitoringFunction.class);
+
+ /** The parallelism of the downstream readers. */
+ private final int readerParallelism;
+
+ /** The interval between consecutive path scans. */
+ private final long interval;
+
+ private final HiveShim hiveShim;
+
+ private final JobConfWrapper conf;
+
+ private final ObjectPath tablePath;
+
+ private final List<String> partitionKeys;
+
+ private final String[] fieldNames;
+
+ private final DataType[] fieldTypes;
+
+ private final long startupTimestampMillis;
+
+ private final PartitionStrategyFactory partStrategyFactory;
+
+ private volatile boolean isRunning = true;
+
+ /** The maximum partition read time seen so far. */
+ private volatile long currentReadTime;
+
+ private transient PartitionStrategy strategy;
+
+ private transient Object checkpointLock;
+
+ private transient ListState<Long> currReadTimeState;
+
+ private transient ListState<List<List<String>>> distinctPartsState;
+
+ private transient IMetaStoreClient client;
+
+ private transient Properties tableProps;
+
+ private transient String defaultPartitionName;
+
+ private transient Set<List<String>> distinctPartitions;
+
+ public HiveContinuousMonitoringFunction(
+ HiveShim hiveShim,
+ JobConf conf,
+ ObjectPath tablePath,
+ CatalogTable catalogTable,
+ long startupTimestampMillis,
+ PartitionStrategyFactory partStrategyFactory,
+ int readerParallelism,
+ long interval) {
+ this.hiveShim = hiveShim;
+ this.conf = new JobConfWrapper(conf);
+ this.tablePath = tablePath;
+ this.partitionKeys = catalogTable.getPartitionKeys();
+ this.fieldNames = catalogTable.getSchema().getFieldNames();
+ this.fieldTypes = catalogTable.getSchema().getFieldDataTypes();
+ this.startupTimestampMillis = startupTimestampMillis;
+ this.partStrategyFactory = partStrategyFactory;
+
+ this.interval = interval;
+ this.readerParallelism = Math.max(readerParallelism, 1);
+ this.currentReadTime = Long.MIN_VALUE;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.strategy = partStrategyFactory.createStrategy(getRuntimeContext().getUserCodeClassLoader());
+
+ this.currReadTimeState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ LongSerializer.INSTANCE
+ )
+ );
+ this.distinctPartsState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
+ )
+ );
+
+ this.client = this.hiveShim.getHiveMetastoreClient(new HiveConf(conf.conf(), HiveConf.class));
+
+ Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
+ this.tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, hiveTable);
+ this.defaultPartitionName = conf.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+ HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
+
+ this.distinctPartitions = new HashSet<>();
+ if (context.isRestored()) {
+ LOG.info("Restoring state for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.currReadTimeState.get().iterator().next();
+ this.distinctPartitions.addAll(this.distinctPartsState.get().iterator().next());
+ } else {
+ LOG.info("No state to restore for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.startupTimestampMillis;
+ }
+ }
+
+ @Override
+ public void run(SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ checkpointLock = context.getCheckpointLock();
+ while (isRunning) {
+ synchronized (checkpointLock) {
+ monitorAndForwardSplits(context);
+ }
+ Thread.sleep(interval);
+ }
+ }
+
+ private void monitorAndForwardSplits(
+ SourceContext<TimestampedHiveInputSplit> context) throws IOException, TException {
+ assert (Thread.holdsLock(checkpointLock));
+
+ List<Partition> partitions = client.listPartitionsByFilter(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ strategy.generateFetchFilter(partitionKeys, currentReadTime),
+ (short) -1);
+
+ if (partitions.isEmpty()) {
+ return;
+ }
+
+ long maxTime = Long.MIN_VALUE;
+ for (Partition partition : partitions) {
Review comment:
The java doc mentions the splits should be forwarded in a specific order. But it seems we're not sorting the partitions here?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4223a26640c7243a61135e803aa1dd21888114a5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683)
* 42953325bb1d6a447ef53e020d3cc311a8c986c6 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
}, {
"hash" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"triggerType" : "PUSH"
}, {
"hash" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929",
"triggerID" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"triggerType" : "PUSH"
}, {
"hash" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937",
"triggerID" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3ac79a41395792dfecefc87a19c58468e91ede8d UNKNOWN
* b3802f894dcdd7226316583e66b37a662c3e5bdb Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929)
* 6545efc0050c0d2ae84755199cf3b5fd55e82dca Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
}, {
"hash" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"triggerType" : "PUSH"
}, {
"hash" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929",
"triggerID" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"triggerType" : "PUSH"
}, {
"hash" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937",
"triggerID" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 44df52a0ff62e1b4ec65efd6a14a8b07d7712e75 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872)
* 3ac79a41395792dfecefc87a19c58468e91ede8d UNKNOWN
* b3802f894dcdd7226316583e66b37a662c3e5bdb Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929)
* 6545efc0050c0d2ae84755199cf3b5fd55e82dca Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422753019
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.flink.table.filesystem;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import java.time.Duration;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds configuration constants used by filesystem(Including hive) connector.
+ */
+public class FileSystemOptions {
+
+ public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
+ key("streaming-source.enable")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Enable streaming source or not.");
+
+ public static final ConfigOption<Duration> STREAMING_SOURCE_MONITOR_INTERVAL =
+ key("streaming-source.monitor-interval")
+ .durationType()
+ .defaultValue(Duration.ofMillis(1))
+ .withDescription("The minimum interval allowed between consecutive partition/file discovery.");
Review comment:
`Time interval for consecutively monitoring partition/file.`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422462637
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,331 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.connectors.hive.read.PartitionFetcher.PartitionFetcherFactory;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+import static org.apache.flink.table.filesystem.PartitionPathUtils.escapePathName;
+
+/**
+ * This is the single (non-parallel) monitoring task which takes a {@link HiveTableInputFormat},
+ * it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring partitions of hive meta store.</li>
+ * <li>Deciding which partitions should be further read and processed.</li>
+ * <li>Creating the {@link HiveTableInputSplit splits} corresponding to those partitions.</li>
+ * <li>Assigning them to downstream tasks for further processing.</li>
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ContinuousFileReaderOperator}
+ * which can have parallelism greater than one.
+ *
+ * <p><b>IMPORTANT NOTE: </b> Splits are forwarded downstream for reading in ascending partition time order,
+ * based on the partition time of the partitions they belong to.
+ */
+public class HiveContinuousMonitoringFunction
+ extends RichSourceFunction<TimestampedHiveInputSplit>
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveContinuousMonitoringFunction.class);
+
+ /** The parallelism of the downstream readers. */
+ private final int readerParallelism;
+
+ /** The interval between consecutive path scans. */
+ private final long interval;
+
+ private final HiveShim hiveShim;
+
+ private final JobConfWrapper conf;
+
+ private final ObjectPath tablePath;
+
+ private final List<String> partitionKeys;
+
+ private final String[] fieldNames;
+
+ private final DataType[] fieldTypes;
+
+ private final String startupPartition;
+
+ private final PartitionFetcherFactory fetcherFactory;
+
+ private volatile boolean isRunning = true;
+
+ /** The maximum partition read time seen so far. */
+ private volatile long currentReadTime;
+
+ private transient PartitionFetcher.Context fetcherContext;
+
+ private transient PartitionFetcher fetcher;
+
+ private transient Object checkpointLock;
+
+ private transient ListState<Long> currReadTimeState;
+
+ private transient ListState<List<List<String>>> distinctPartsState;
+
+ private transient IMetaStoreClient client;
+
+ private transient Properties tableProps;
+
+ private transient String defaultPartitionName;
+
+ private transient Set<List<String>> distinctPartitions;
+
+ public HiveContinuousMonitoringFunction(
+ HiveShim hiveShim,
+ JobConf conf,
+ ObjectPath tablePath,
+ CatalogTable catalogTable,
+ String startupPartition,
+ PartitionFetcherFactory fetcherFactory,
+ int readerParallelism,
+ long interval) {
+ this.hiveShim = hiveShim;
+ this.conf = new JobConfWrapper(conf);
+ this.tablePath = tablePath;
+ this.partitionKeys = catalogTable.getPartitionKeys();
+ this.fieldNames = catalogTable.getSchema().getFieldNames();
+ this.fieldTypes = catalogTable.getSchema().getFieldDataTypes();
+ this.startupPartition = startupPartition;
+ this.fetcherFactory = fetcherFactory;
+
+ this.interval = interval;
+ this.readerParallelism = Math.max(readerParallelism, 1);
+ this.currentReadTime = 0;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.currReadTimeState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ LongSerializer.INSTANCE
+ )
+ );
+ this.distinctPartsState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
+ )
+ );
+
+ this.client = this.hiveShim.getHiveMetastoreClient(new HiveConf(conf.conf(), HiveConf.class));
+
+ Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
+ this.tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, hiveTable);
+ this.defaultPartitionName = conf.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+ HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
+
+ this.fetcher = fetcherFactory.createStrategy(getRuntimeContext().getUserCodeClassLoader());
+
+ Path location = new Path(hiveTable.getSd().getLocation());
+ FileSystem fs = location.getFileSystem(conf.conf());
+ this.fetcherContext = new PartitionFetcher.Context() {
+
+ @Override
+ public List<String> partitionKeys() {
+ return partitionKeys;
+ }
+
+ @Override
+ public Partition getPartition(String escapeName) throws TException {
+ return client.getPartition(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ escapeName);
+ }
+
+ @Override
+ public List<Partition> listPartitionsByFilter(String filter) throws TException {
+ return client.listPartitionsByFilter(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ filter,
+ (short) -1);
+ }
+
+ @Override
+ public FileSystem fileSystem() {
+ return fs;
+ }
+
+ @Override
+ public Path tableLocation() {
+ return new Path(hiveTable.getSd().getLocation());
+ }
+ };
+
+ this.distinctPartitions = new HashSet<>();
+ if (context.isRestored()) {
+ LOG.info("Restoring state for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.currReadTimeState.get().iterator().next();
+ this.distinctPartitions.addAll(this.distinctPartsState.get().iterator().next());
+ } else {
+ LOG.info("No state to restore for the {}.", getClass().getSimpleName());
+ if (startupPartition != null) {
+ this.currentReadTime = fetcher.extractTimestamp(
+ fetcherContext,
+ client.getPartition(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ escapePathName(startupPartition)));
+ }
+ }
+ }
+
+ @Override
+ public void run(SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ checkpointLock = context.getCheckpointLock();
+ while (isRunning) {
+ synchronized (checkpointLock) {
+ monitorAndForwardSplits(context);
+ }
+ Thread.sleep(interval);
+ }
+ }
+
+ private void monitorAndForwardSplits(
+ SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ assert (Thread.holdsLock(checkpointLock));
+
+ List<Partition> partitions = fetcher.fetchPartitions(fetcherContext, currentReadTime);
+
+ if (partitions.isEmpty()) {
+ return;
+ }
+
+ long maxTimestamp = Long.MIN_VALUE;
+ Set<List<String>> nextDistinctParts = new HashSet<>();
+ for (Partition partition : partitions) {
+ List<String> partSpec = partition.getValues();
+ if (!this.distinctPartitions.contains(partSpec)) {
+ this.distinctPartitions.add(partition.getValues());
+ long timestamp = this.fetcher.extractTimestamp(fetcherContext, partition);
+ if (timestamp > currentReadTime) {
+ nextDistinctParts.add(partSpec);
+ }
+ if (timestamp > maxTimestamp) {
+ maxTimestamp = timestamp;
+ }
+ HiveTableInputSplit[] splits = HiveTableInputFormat.createInputSplits(
+ this.readerParallelism,
+ Collections.singletonList(toHiveTablePartition(partition)),
+ this.conf.conf());
+ for (HiveTableInputSplit split : splits) {
+ context.collect(new TimestampedHiveInputSplit(timestamp, split));
+ }
+ }
+ }
+
+ if (maxTimestamp > currentReadTime) {
Review comment:
> we will lost part of the data to consume
No, these partitions will be all emitted to downstream, there is no checkpoint in them.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666)
* a9a2c659b1693752361aaf3cf6f67e8251032983 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4223a26640c7243a61135e803aa1dd21888114a5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422752473
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/TimestampedHiveInputSplit.java
##########
@@ -0,0 +1,232 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.streaming.api.functions.source.TimestampedInputSplit;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.lang.reflect.Type;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * A {@link HiveTableInputSplit} with {@link TimestampedInputSplit}.
+ * Kryo serializer can not deal with hadoop split, need specific type information factory.
+ */
+@TypeInfo(TimestampedHiveInputSplit.SplitTypeInfoFactory.class)
+public class TimestampedHiveInputSplit extends HiveTableInputSplit implements TimestampedInputSplit {
+
+ private static final long serialVersionUID = 1L;
+
+ /** The modification time of the file this split belongs to. */
+ private final long modificationTime;
+
+ /**
+ * The state of the split. This information is used when
+ * restoring from a checkpoint and allows to resume reading the
+ * underlying file from the point we left off.
+ * */
+ private Serializable splitState;
+
+ public TimestampedHiveInputSplit(
+ long modificationTime,
+ HiveTableInputSplit split) {
+ super(
+ split.getSplitNumber(),
+ split.getHadoopInputSplit(),
+ split.getJobConf(),
+ split.getHiveTablePartition());
+ this.modificationTime = modificationTime;
+ }
+
+ @Override
+ public void setSplitState(Serializable state) {
+ this.splitState = state;
+ }
+
+ @Override
+ public Serializable getSplitState() {
+ return this.splitState;
+ }
+
+ @Override
+ public long getModificationTime() {
+ return modificationTime;
+ }
+
+ @Override
+ public int compareTo(TimestampedInputSplit o) {
+ TimestampedHiveInputSplit split = (TimestampedHiveInputSplit) o;
+ int modTimeComp = Long.compare(this.modificationTime, split.modificationTime);
+ if (modTimeComp != 0L) {
+ return modTimeComp;
+ }
+
+ int pathComp = this.hiveTablePartition.getStorageDescriptor().compareTo(
Review comment:
It is a short cut for comparing. If the sd is same, `SplitNumber` must be different.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422757793
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/TimestampedHiveInputSplit.java
##########
@@ -0,0 +1,232 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.streaming.api.functions.source.TimestampedInputSplit;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.lang.reflect.Type;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * A {@link HiveTableInputSplit} with {@link TimestampedInputSplit}.
+ * Kryo serializer can not deal with hadoop split, need specific type information factory.
+ */
+@TypeInfo(TimestampedHiveInputSplit.SplitTypeInfoFactory.class)
+public class TimestampedHiveInputSplit extends HiveTableInputSplit implements TimestampedInputSplit {
+
+ private static final long serialVersionUID = 1L;
+
+ /** The modification time of the file this split belongs to. */
+ private final long modificationTime;
+
+ /**
+ * The state of the split. This information is used when
+ * restoring from a checkpoint and allows to resume reading the
+ * underlying file from the point we left off.
+ * */
+ private Serializable splitState;
+
+ public TimestampedHiveInputSplit(
+ long modificationTime,
+ HiveTableInputSplit split) {
+ super(
+ split.getSplitNumber(),
+ split.getHadoopInputSplit(),
+ split.getJobConf(),
+ split.getHiveTablePartition());
+ this.modificationTime = modificationTime;
+ }
+
+ @Override
+ public void setSplitState(Serializable state) {
+ this.splitState = state;
+ }
+
+ @Override
+ public Serializable getSplitState() {
+ return this.splitState;
+ }
+
+ @Override
+ public long getModificationTime() {
+ return modificationTime;
+ }
+
+ @Override
+ public int compareTo(TimestampedInputSplit o) {
+ TimestampedHiveInputSplit split = (TimestampedHiveInputSplit) o;
+ int modTimeComp = Long.compare(this.modificationTime, split.modificationTime);
+ if (modTimeComp != 0L) {
+ return modTimeComp;
+ }
+
+ int pathComp = this.hiveTablePartition.getStorageDescriptor().compareTo(
+ split.hiveTablePartition.getStorageDescriptor());
+
+ return pathComp != 0 ? pathComp :
+ this.getSplitNumber() - o.getSplitNumber();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
Review comment:
`TimestampedHiveInputSplit` is only used in `ContinuousFileReaderOperator`. I will add comments.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
}, {
"hash" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 44df52a0ff62e1b4ec65efd6a14a8b07d7712e75 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872)
* 3ac79a41395792dfecefc87a19c58468e91ede8d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422462527
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,331 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.connectors.hive.read.PartitionFetcher.PartitionFetcherFactory;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+import static org.apache.flink.table.filesystem.PartitionPathUtils.escapePathName;
+
+/**
+ * This is the single (non-parallel) monitoring task which takes a {@link HiveTableInputFormat},
+ * it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring partitions of hive meta store.</li>
+ * <li>Deciding which partitions should be further read and processed.</li>
+ * <li>Creating the {@link HiveTableInputSplit splits} corresponding to those partitions.</li>
+ * <li>Assigning them to downstream tasks for further processing.</li>
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ContinuousFileReaderOperator}
+ * which can have parallelism greater than one.
+ *
+ * <p><b>IMPORTANT NOTE: </b> Splits are forwarded downstream for reading in ascending partition time order,
+ * based on the partition time of the partitions they belong to.
+ */
+public class HiveContinuousMonitoringFunction
+ extends RichSourceFunction<TimestampedHiveInputSplit>
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveContinuousMonitoringFunction.class);
+
+ /** The parallelism of the downstream readers. */
+ private final int readerParallelism;
+
+ /** The interval between consecutive path scans. */
+ private final long interval;
+
+ private final HiveShim hiveShim;
+
+ private final JobConfWrapper conf;
+
+ private final ObjectPath tablePath;
+
+ private final List<String> partitionKeys;
+
+ private final String[] fieldNames;
+
+ private final DataType[] fieldTypes;
+
+ private final String startupPartition;
+
+ private final PartitionFetcherFactory fetcherFactory;
+
+ private volatile boolean isRunning = true;
+
+ /** The maximum partition read time seen so far. */
+ private volatile long currentReadTime;
+
+ private transient PartitionFetcher.Context fetcherContext;
+
+ private transient PartitionFetcher fetcher;
+
+ private transient Object checkpointLock;
+
+ private transient ListState<Long> currReadTimeState;
+
+ private transient ListState<List<List<String>>> distinctPartsState;
+
+ private transient IMetaStoreClient client;
+
+ private transient Properties tableProps;
+
+ private transient String defaultPartitionName;
+
+ private transient Set<List<String>> distinctPartitions;
+
+ public HiveContinuousMonitoringFunction(
+ HiveShim hiveShim,
+ JobConf conf,
+ ObjectPath tablePath,
+ CatalogTable catalogTable,
+ String startupPartition,
+ PartitionFetcherFactory fetcherFactory,
+ int readerParallelism,
+ long interval) {
+ this.hiveShim = hiveShim;
+ this.conf = new JobConfWrapper(conf);
+ this.tablePath = tablePath;
+ this.partitionKeys = catalogTable.getPartitionKeys();
+ this.fieldNames = catalogTable.getSchema().getFieldNames();
+ this.fieldTypes = catalogTable.getSchema().getFieldDataTypes();
+ this.startupPartition = startupPartition;
+ this.fetcherFactory = fetcherFactory;
+
+ this.interval = interval;
+ this.readerParallelism = Math.max(readerParallelism, 1);
+ this.currentReadTime = 0;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.currReadTimeState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ LongSerializer.INSTANCE
+ )
+ );
+ this.distinctPartsState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
+ )
+ );
+
+ this.client = this.hiveShim.getHiveMetastoreClient(new HiveConf(conf.conf(), HiveConf.class));
+
+ Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
+ this.tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, hiveTable);
+ this.defaultPartitionName = conf.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+ HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
+
+ this.fetcher = fetcherFactory.createStrategy(getRuntimeContext().getUserCodeClassLoader());
+
+ Path location = new Path(hiveTable.getSd().getLocation());
+ FileSystem fs = location.getFileSystem(conf.conf());
+ this.fetcherContext = new PartitionFetcher.Context() {
+
+ @Override
+ public List<String> partitionKeys() {
+ return partitionKeys;
+ }
+
+ @Override
+ public Partition getPartition(String escapeName) throws TException {
+ return client.getPartition(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ escapeName);
+ }
+
+ @Override
+ public List<Partition> listPartitionsByFilter(String filter) throws TException {
+ return client.listPartitionsByFilter(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ filter,
+ (short) -1);
+ }
+
+ @Override
+ public FileSystem fileSystem() {
+ return fs;
+ }
+
+ @Override
+ public Path tableLocation() {
+ return new Path(hiveTable.getSd().getLocation());
+ }
+ };
+
+ this.distinctPartitions = new HashSet<>();
+ if (context.isRestored()) {
+ LOG.info("Restoring state for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.currReadTimeState.get().iterator().next();
+ this.distinctPartitions.addAll(this.distinctPartsState.get().iterator().next());
+ } else {
+ LOG.info("No state to restore for the {}.", getClass().getSimpleName());
+ if (startupPartition != null) {
+ this.currentReadTime = fetcher.extractTimestamp(
+ fetcherContext,
+ client.getPartition(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ escapePathName(startupPartition)));
+ }
+ }
+ }
+
+ @Override
+ public void run(SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ checkpointLock = context.getCheckpointLock();
+ while (isRunning) {
+ synchronized (checkpointLock) {
+ monitorAndForwardSplits(context);
+ }
+ Thread.sleep(interval);
+ }
+ }
+
+ private void monitorAndForwardSplits(
+ SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ assert (Thread.holdsLock(checkpointLock));
+
+ List<Partition> partitions = fetcher.fetchPartitions(fetcherContext, currentReadTime);
+
+ if (partitions.isEmpty()) {
+ return;
+ }
+
+ long maxTimestamp = Long.MIN_VALUE;
+ Set<List<String>> nextDistinctParts = new HashSet<>();
+ for (Partition partition : partitions) {
+ List<String> partSpec = partition.getValues();
+ if (!this.distinctPartitions.contains(partSpec)) {
+ this.distinctPartitions.add(partition.getValues());
+ long timestamp = this.fetcher.extractTimestamp(fetcherContext, partition);
+ if (timestamp > currentReadTime) {
+ nextDistinctParts.add(partSpec);
+ }
+ if (timestamp > maxTimestamp) {
+ maxTimestamp = timestamp;
+ }
+ HiveTableInputSplit[] splits = HiveTableInputFormat.createInputSplits(
+ this.readerParallelism,
+ Collections.singletonList(toHiveTablePartition(partition)),
+ this.conf.conf());
+ for (HiveTableInputSplit split : splits) {
+ context.collect(new TimestampedHiveInputSplit(timestamp, split));
+ }
+ }
+ }
+
+ if (maxTimestamp > currentReadTime) {
Review comment:
First, will sort this list.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] openinx commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422124325
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
##########
@@ -181,6 +214,47 @@ public boolean isBounded() {
return source.name(explainSource());
}
+ private DataStream<RowData> createStreamSource(
+ StreamExecutionEnvironment execEnv,
+ TypeInformation<RowData> typeInfo,
+ HiveTableInputFormat inputFormat) {
+ final Map<String, String> properties = catalogTable.getProperties();
+ PartitionFetcherFactory strategyFactory = cl -> PartitionFetcher.createStrategy(
+ properties.get(HIVE_STREAMING_SOURCE_PARTITION_STRATEGY.key()),
+ properties.get(HIVE_STREAMING_SOURCE_PARTITION_STRATEGY_CLASS.key()),
+ cl);
+
+ String monitorIntervalStr = properties.get(HIVE_STREAMING_SOURCE_MONITOR_INTERVAL.key());
+ Duration monitorInterval = monitorIntervalStr != null ?
+ TimeUtils.parseDuration(monitorIntervalStr) :
+ HIVE_STREAMING_SOURCE_MONITOR_INTERVAL.defaultValue();
+
+ HiveContinuousMonitoringFunction monitoringFunction = new HiveContinuousMonitoringFunction(
+ hiveShim,
+ jobConf,
+ tablePath,
+ catalogTable,
+ getStartupPartition(),
+ strategyFactory,
+ execEnv.getParallelism(),
+ monitorInterval.toMillis());
+
+ ContinuousFileReaderOperatorFactory<RowData, TimestampedHiveInputSplit> factory =
+ new ContinuousFileReaderOperatorFactory<>(inputFormat);
Review comment:
Besides seems we don't do any split inside the file which may introduce the unbalanced data consuming in the upstream operator ? Iceberg seems handle this well because it will split the larger data file into small balanced tasks and dispatch to the executing task..
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
##########
@@ -181,6 +214,47 @@ public boolean isBounded() {
return source.name(explainSource());
}
+ private DataStream<RowData> createStreamSource(
+ StreamExecutionEnvironment execEnv,
+ TypeInformation<RowData> typeInfo,
+ HiveTableInputFormat inputFormat) {
+ final Map<String, String> properties = catalogTable.getProperties();
+ PartitionFetcherFactory strategyFactory = cl -> PartitionFetcher.createStrategy(
+ properties.get(HIVE_STREAMING_SOURCE_PARTITION_STRATEGY.key()),
+ properties.get(HIVE_STREAMING_SOURCE_PARTITION_STRATEGY_CLASS.key()),
+ cl);
+
+ String monitorIntervalStr = properties.get(HIVE_STREAMING_SOURCE_MONITOR_INTERVAL.key());
+ Duration monitorInterval = monitorIntervalStr != null ?
+ TimeUtils.parseDuration(monitorIntervalStr) :
+ HIVE_STREAMING_SOURCE_MONITOR_INTERVAL.defaultValue();
+
+ HiveContinuousMonitoringFunction monitoringFunction = new HiveContinuousMonitoringFunction(
+ hiveShim,
+ jobConf,
+ tablePath,
+ catalogTable,
+ getStartupPartition(),
+ strategyFactory,
+ execEnv.getParallelism(),
+ monitorInterval.toMillis());
+
+ ContinuousFileReaderOperatorFactory<RowData, TimestampedHiveInputSplit> factory =
+ new ContinuousFileReaderOperatorFactory<>(inputFormat);
Review comment:
I've one question here, what's the parallelism of the ContinuousFileReaderOperator will be ? should we use the default parallelism in ExecutionConfig ?
OK, seems the InputFormat will generate the expected parallelism...
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
##########
@@ -150,6 +164,25 @@ public boolean isBounded() {
allHivePartitions,
flinkConf.get(HiveOptions.TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER));
+ if (isStreamingSource()) {
+ if (catalogTable.getPartitionKeys().isEmpty()) {
+ throw new UnsupportedOperationException(
+ "Non-partition table not support streaming read now.");
Review comment:
not -> does not .
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,331 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.connectors.hive.read.PartitionFetcher.PartitionFetcherFactory;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
Review comment:
Seems the useless imported class can be removed
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,331 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.connectors.hive.read.PartitionFetcher.PartitionFetcherFactory;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+import static org.apache.flink.table.filesystem.PartitionPathUtils.escapePathName;
+
+/**
+ * This is the single (non-parallel) monitoring task which takes a {@link HiveTableInputFormat},
+ * it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring partitions of hive meta store.</li>
+ * <li>Deciding which partitions should be further read and processed.</li>
+ * <li>Creating the {@link HiveTableInputSplit splits} corresponding to those partitions.</li>
+ * <li>Assigning them to downstream tasks for further processing.</li>
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ContinuousFileReaderOperator}
+ * which can have parallelism greater than one.
+ *
+ * <p><b>IMPORTANT NOTE: </b> Splits are forwarded downstream for reading in ascending partition time order,
+ * based on the partition time of the partitions they belong to.
+ */
+public class HiveContinuousMonitoringFunction
+ extends RichSourceFunction<TimestampedHiveInputSplit>
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveContinuousMonitoringFunction.class);
+
+ /** The parallelism of the downstream readers. */
+ private final int readerParallelism;
+
+ /** The interval between consecutive path scans. */
+ private final long interval;
+
+ private final HiveShim hiveShim;
+
+ private final JobConfWrapper conf;
+
+ private final ObjectPath tablePath;
+
+ private final List<String> partitionKeys;
+
+ private final String[] fieldNames;
+
+ private final DataType[] fieldTypes;
+
+ private final String startupPartition;
+
+ private final PartitionFetcherFactory fetcherFactory;
+
+ private volatile boolean isRunning = true;
+
+ /** The maximum partition read time seen so far. */
+ private volatile long currentReadTime;
+
+ private transient PartitionFetcher.Context fetcherContext;
+
+ private transient PartitionFetcher fetcher;
+
+ private transient Object checkpointLock;
+
+ private transient ListState<Long> currReadTimeState;
+
+ private transient ListState<List<List<String>>> distinctPartsState;
+
+ private transient IMetaStoreClient client;
+
+ private transient Properties tableProps;
+
+ private transient String defaultPartitionName;
+
+ private transient Set<List<String>> distinctPartitions;
+
+ public HiveContinuousMonitoringFunction(
+ HiveShim hiveShim,
+ JobConf conf,
+ ObjectPath tablePath,
+ CatalogTable catalogTable,
+ String startupPartition,
+ PartitionFetcherFactory fetcherFactory,
+ int readerParallelism,
+ long interval) {
+ this.hiveShim = hiveShim;
+ this.conf = new JobConfWrapper(conf);
+ this.tablePath = tablePath;
+ this.partitionKeys = catalogTable.getPartitionKeys();
+ this.fieldNames = catalogTable.getSchema().getFieldNames();
+ this.fieldTypes = catalogTable.getSchema().getFieldDataTypes();
+ this.startupPartition = startupPartition;
+ this.fetcherFactory = fetcherFactory;
+
+ this.interval = interval;
+ this.readerParallelism = Math.max(readerParallelism, 1);
+ this.currentReadTime = 0;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.currReadTimeState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ LongSerializer.INSTANCE
+ )
+ );
+ this.distinctPartsState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
+ )
+ );
+
+ this.client = this.hiveShim.getHiveMetastoreClient(new HiveConf(conf.conf(), HiveConf.class));
+
+ Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
+ this.tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, hiveTable);
+ this.defaultPartitionName = conf.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+ HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
+
+ this.fetcher = fetcherFactory.createStrategy(getRuntimeContext().getUserCodeClassLoader());
+
+ Path location = new Path(hiveTable.getSd().getLocation());
+ FileSystem fs = location.getFileSystem(conf.conf());
+ this.fetcherContext = new PartitionFetcher.Context() {
+
+ @Override
+ public List<String> partitionKeys() {
+ return partitionKeys;
+ }
+
+ @Override
+ public Partition getPartition(String escapeName) throws TException {
+ return client.getPartition(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ escapeName);
+ }
+
+ @Override
+ public List<Partition> listPartitionsByFilter(String filter) throws TException {
+ return client.listPartitionsByFilter(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ filter,
+ (short) -1);
+ }
+
+ @Override
+ public FileSystem fileSystem() {
+ return fs;
+ }
+
+ @Override
+ public Path tableLocation() {
+ return new Path(hiveTable.getSd().getLocation());
+ }
+ };
+
+ this.distinctPartitions = new HashSet<>();
+ if (context.isRestored()) {
+ LOG.info("Restoring state for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.currReadTimeState.get().iterator().next();
+ this.distinctPartitions.addAll(this.distinctPartsState.get().iterator().next());
+ } else {
+ LOG.info("No state to restore for the {}.", getClass().getSimpleName());
+ if (startupPartition != null) {
+ this.currentReadTime = fetcher.extractTimestamp(
+ fetcherContext,
+ client.getPartition(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ escapePathName(startupPartition)));
+ }
+ }
+ }
+
+ @Override
+ public void run(SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ checkpointLock = context.getCheckpointLock();
+ while (isRunning) {
+ synchronized (checkpointLock) {
+ monitorAndForwardSplits(context);
+ }
+ Thread.sleep(interval);
+ }
+ }
+
+ private void monitorAndForwardSplits(
+ SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ assert (Thread.holdsLock(checkpointLock));
+
+ List<Partition> partitions = fetcher.fetchPartitions(fetcherContext, currentReadTime);
+
+ if (partitions.isEmpty()) {
+ return;
+ }
+
+ long maxTimestamp = Long.MIN_VALUE;
+ Set<List<String>> nextDistinctParts = new HashSet<>();
+ for (Partition partition : partitions) {
+ List<String> partSpec = partition.getValues();
+ if (!this.distinctPartitions.contains(partSpec)) {
+ this.distinctPartitions.add(partition.getValues());
+ long timestamp = this.fetcher.extractTimestamp(fetcherContext, partition);
+ if (timestamp > currentReadTime) {
+ nextDistinctParts.add(partSpec);
+ }
+ if (timestamp > maxTimestamp) {
+ maxTimestamp = timestamp;
+ }
+ HiveTableInputSplit[] splits = HiveTableInputFormat.createInputSplits(
+ this.readerParallelism,
+ Collections.singletonList(toHiveTablePartition(partition)),
+ this.conf.conf());
+ for (HiveTableInputSplit split : splits) {
+ context.collect(new TimestampedHiveInputSplit(timestamp, split));
+ }
+ }
+ }
+
+ if (maxTimestamp > currentReadTime) {
Review comment:
Just ask the similar question as @lirui-apache said, if the partitions is not sorted by timestamp ascending, then it seems have problems, say we have partitions with timestamp like `[(p0, 5), (p1, 1), (p2, 3)]` ( the first part of (p0, 5) is partition name and the second is timestamp ), now we have emitted the timestamp=5 successfully, then we will update the currentReadTime = 5 I think. Once the operator restored, we will just start from the timestamp = 5 to read the following partitions...it mistakenly skipped other timestamps such as 1 and 3 , finally we will lost part of the data to consume....
Pls correct me if I'm wrong.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666)
* a9a2c659b1693752361aaf3cf6f67e8251032983 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
}, {
"hash" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"triggerType" : "PUSH"
}, {
"hash" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 44df52a0ff62e1b4ec65efd6a14a8b07d7712e75 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872)
* 3ac79a41395792dfecefc87a19c58468e91ede8d UNKNOWN
* b3802f894dcdd7226316583e66b37a662c3e5bdb UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
}, {
"hash" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"triggerType" : "PUSH"
}, {
"hash" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929",
"triggerID" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"triggerType" : "PUSH"
}, {
"hash" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937",
"triggerID" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"triggerType" : "PUSH"
}, {
"hash" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=946",
"triggerID" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"triggerType" : "PUSH"
}, {
"hash" : "bffea7d6cacc503cc630c8b3ddf77f4093554e84",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=967",
"triggerID" : "bffea7d6cacc503cc630c8b3ddf77f4093554e84",
"triggerType" : "PUSH"
}, {
"hash" : "48017434485d6059dafddc8617bf7f75fa7b3892",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=982",
"triggerID" : "48017434485d6059dafddc8617bf7f75fa7b3892",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3ac79a41395792dfecefc87a19c58468e91ede8d UNKNOWN
* bffea7d6cacc503cc630c8b3ddf77f4093554e84 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=967)
* 48017434485d6059dafddc8617bf7f75fa7b3892 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=982)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
}, {
"hash" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"triggerType" : "PUSH"
}, {
"hash" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929",
"triggerID" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"triggerType" : "PUSH"
}, {
"hash" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 44df52a0ff62e1b4ec65efd6a14a8b07d7712e75 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872)
* 3ac79a41395792dfecefc87a19c58468e91ede8d UNKNOWN
* b3802f894dcdd7226316583e66b37a662c3e5bdb Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929)
* 6545efc0050c0d2ae84755199cf3b5fd55e82dca UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624626103
@flinkbot run azure
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
}, {
"hash" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"triggerType" : "PUSH"
}, {
"hash" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929",
"triggerID" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"triggerType" : "PUSH"
}, {
"hash" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937",
"triggerID" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"triggerType" : "PUSH"
}, {
"hash" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=946",
"triggerID" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"triggerType" : "PUSH"
}, {
"hash" : "bffea7d6cacc503cc630c8b3ddf77f4093554e84",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=967",
"triggerID" : "bffea7d6cacc503cc630c8b3ddf77f4093554e84",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3ac79a41395792dfecefc87a19c58468e91ede8d UNKNOWN
* 6545efc0050c0d2ae84755199cf3b5fd55e82dca Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937)
* 006610cb59c3665b510485c6d6ad883d91fc3e6d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=946)
* bffea7d6cacc503cc630c8b3ddf77f4093554e84 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=967)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
}, {
"hash" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"triggerType" : "PUSH"
}, {
"hash" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929",
"triggerID" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"triggerType" : "PUSH"
}, {
"hash" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937",
"triggerID" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"triggerType" : "PUSH"
}, {
"hash" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3ac79a41395792dfecefc87a19c58468e91ede8d UNKNOWN
* b3802f894dcdd7226316583e66b37a662c3e5bdb Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929)
* 6545efc0050c0d2ae84755199cf3b5fd55e82dca Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937)
* 006610cb59c3665b510485c6d6ad883d91fc3e6d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422752654
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/TimestampedHiveInputSplit.java
##########
@@ -0,0 +1,232 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.streaming.api.functions.source.TimestampedInputSplit;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.lang.reflect.Type;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * A {@link HiveTableInputSplit} with {@link TimestampedInputSplit}.
+ * Kryo serializer can not deal with hadoop split, need specific type information factory.
+ */
+@TypeInfo(TimestampedHiveInputSplit.SplitTypeInfoFactory.class)
+public class TimestampedHiveInputSplit extends HiveTableInputSplit implements TimestampedInputSplit {
+
+ private static final long serialVersionUID = 1L;
+
+ /** The modification time of the file this split belongs to. */
+ private final long modificationTime;
+
+ /**
+ * The state of the split. This information is used when
+ * restoring from a checkpoint and allows to resume reading the
+ * underlying file from the point we left off.
+ * */
+ private Serializable splitState;
+
+ public TimestampedHiveInputSplit(
+ long modificationTime,
+ HiveTableInputSplit split) {
+ super(
+ split.getSplitNumber(),
+ split.getHadoopInputSplit(),
+ split.getJobConf(),
+ split.getHiveTablePartition());
+ this.modificationTime = modificationTime;
+ }
+
+ @Override
+ public void setSplitState(Serializable state) {
+ this.splitState = state;
+ }
+
+ @Override
+ public Serializable getSplitState() {
+ return this.splitState;
+ }
+
+ @Override
+ public long getModificationTime() {
+ return modificationTime;
+ }
+
+ @Override
+ public int compareTo(TimestampedInputSplit o) {
+ TimestampedHiveInputSplit split = (TimestampedHiveInputSplit) o;
+ int modTimeComp = Long.compare(this.modificationTime, split.modificationTime);
+ if (modTimeComp != 0L) {
+ return modTimeComp;
+ }
+
+ int pathComp = this.hiveTablePartition.getStorageDescriptor().compareTo(
+ split.hiveTablePartition.getStorageDescriptor());
+
+ return pathComp != 0 ? pathComp :
+ this.getSplitNumber() - o.getSplitNumber();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
Review comment:
Yes, we can create the case like you said, but in read world, it should not happen.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422753438
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.flink.table.filesystem;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import java.time.Duration;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds configuration constants used by filesystem(Including hive) connector.
+ */
+public class FileSystemOptions {
+
+ public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
+ key("streaming-source.enable")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Enable streaming source or not.");
+
+ public static final ConfigOption<Duration> STREAMING_SOURCE_MONITOR_INTERVAL =
+ key("streaming-source.monitor-interval")
+ .durationType()
+ .defaultValue(Duration.ofMillis(1))
+ .withDescription("The minimum interval allowed between consecutive partition/file discovery.");
+
+ public static final ConfigOption<String> STREAMING_SOURCE_CONSUME_ORDER =
+ key("streaming-source.consume-order")
+ .stringType()
+ .defaultValue("create-time")
+ .withDescription("The consume order of streaming source," +
+ " support create-time and partition-time." +
+ " create-time compare partition/file creation time;" +
+ " partition-time compare time represented by partition name.");
+
+ public static final ConfigOption<String> STREAMING_SOURCE_CONSUME_START_OFFSET =
+ key("streaming-source.consume-start-offset")
+ .stringType()
+ .defaultValue("1970-00-00")
+ .withDescription("Start offset for streaming consuming." +
+ " How to parse and compare offsets depends on your order." +
+ " For create-time and partition-time, should be a timestamp string.");
+
+ public static final ConfigOption<String> PARTITION_TIME_EXTRACTOR_TYPE =
+ key("partition.time-extractor.type")
+ .stringType()
+ .defaultValue("timestamp")
+ .withDescription("Time extractor to extract time from partition values." +
Review comment:
Yes, will add comments.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
}, {
"hash" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"triggerType" : "PUSH"
}, {
"hash" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929",
"triggerID" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"triggerType" : "PUSH"
}, {
"hash" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937",
"triggerID" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"triggerType" : "PUSH"
}, {
"hash" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=946",
"triggerID" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3ac79a41395792dfecefc87a19c58468e91ede8d UNKNOWN
* 6545efc0050c0d2ae84755199cf3b5fd55e82dca Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937)
* 006610cb59c3665b510485c6d6ad883d91fc3e6d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=946)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 44df52a0ff62e1b4ec65efd6a14a8b07d7712e75 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lirui-apache commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
lirui-apache commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r421206166
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveTableInputFormat.java
##########
@@ -248,7 +263,11 @@ public RowData nextRecord(RowData reuse) throws IOException {
}
@Override
- public void close() throws IOException {
+ public void close() {
Review comment:
What's the difference between `close` and `closeInputFormat`? And can we have some comments explaining why we don't have to do anything in `close`?
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,358 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.filesystem.PartitionTimeExtractor;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Supplier;
+
+import static org.apache.flink.table.filesystem.TimestampPartTimeExtractor.toLocalDateTime;
+
+/**
+ * This is the single (non-parallel) monitoring task which takes a {@link HiveTableInputFormat},
+ * it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring partitions of hive meta store.</li>
+ * <li>Deciding which partitions should be further read and processed.</li>
+ * <li>Creating the {@link HiveTableInputSplit splits} corresponding to those partitions.</li>
+ * <li>Assigning them to downstream tasks for further processing.</li>
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ContinuousFileReaderOperator}
+ * which can have parallelism greater than one.
+ *
+ * <p><b>IMPORTANT NOTE: </b> Splits are forwarded downstream for reading in ascending partition time order,
+ * based on the partition time of the partitions they belong to.
+ */
+public class HiveContinuousMonitoringFunction
+ extends RichSourceFunction<TimestampedHiveInputSplit>
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveContinuousMonitoringFunction.class);
+
+ /** The parallelism of the downstream readers. */
+ private final int readerParallelism;
+
+ /** The interval between consecutive path scans. */
+ private final long interval;
+
+ private final HiveShim hiveShim;
+
+ private final JobConfWrapper conf;
+
+ private final ObjectPath tablePath;
+
+ private final List<String> partitionKeys;
+
+ private final String[] fieldNames;
+
+ private final DataType[] fieldTypes;
+
+ // consumer variables
+ private final String consumeOrder;
+ private final String consumeOffset;
+
+ // extractor variables
+ private final String extractorType;
+ private final String extractorClass;
+ private final String extractorPattern;
+
+ private volatile boolean isRunning = true;
+
+ /** The maximum partition read time seen so far. */
+ private volatile long currentReadTime;
+
+ private transient PartitionDiscovery.Context context;
+
+ private transient PartitionDiscovery fetcher;
+
+ private transient Object checkpointLock;
+
+ private transient ListState<Long> currReadTimeState;
+
+ private transient ListState<List<List<String>>> distinctPartsState;
+
+ private transient IMetaStoreClient client;
+
+ private transient Properties tableProps;
+
+ private transient String defaultPartitionName;
+
+ private transient Set<List<String>> distinctPartitions;
+
+ public HiveContinuousMonitoringFunction(
+ HiveShim hiveShim,
+ JobConf conf,
+ ObjectPath tablePath,
+ CatalogTable catalogTable,
+ int readerParallelism,
+ String consumeOrder,
+ String consumeOffset,
+ String extractorType,
+ String extractorClass,
+ String extractorPattern,
+ long interval) {
+ this.hiveShim = hiveShim;
+ this.conf = new JobConfWrapper(conf);
+ this.tablePath = tablePath;
+ this.partitionKeys = catalogTable.getPartitionKeys();
+ this.fieldNames = catalogTable.getSchema().getFieldNames();
+ this.fieldTypes = catalogTable.getSchema().getFieldDataTypes();
+ this.consumeOrder = consumeOrder;
+ this.extractorType = extractorType;
+ this.extractorClass = extractorClass;
+ this.extractorPattern = extractorPattern;
+ this.consumeOffset = consumeOffset;
+
+ this.interval = interval;
+ this.readerParallelism = Math.max(readerParallelism, 1);
+ this.currentReadTime = 0;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.currReadTimeState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ LongSerializer.INSTANCE
+ )
+ );
+ this.distinctPartsState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
+ )
+ );
+
+ this.client = this.hiveShim.getHiveMetastoreClient(new HiveConf(conf.conf(), HiveConf.class));
+
+ Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
+ this.tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, hiveTable);
+ this.defaultPartitionName = conf.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+ HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
+
+ PartitionTimeExtractor extractor = PartitionTimeExtractor.create(
+ getRuntimeContext().getUserCodeClassLoader(),
+ extractorType,
+ extractorClass,
+ extractorPattern);
+
+ this.fetcher = new DirectoryMonitorDiscovery();
+
+ Path location = new Path(hiveTable.getSd().getLocation());
+ FileSystem fs = location.getFileSystem(conf.conf());
+ this.context = new PartitionDiscovery.Context() {
+
+ @Override
+ public List<String> partitionKeys() {
+ return partitionKeys;
+ }
+
+ @Override
+ public Partition getPartition(List<String> partValues) throws TException {
+ return client.getPartition(
Review comment:
I think it's possible that a partition folder exists in FS but not in HMS, in which case that folder should be simply ignored instead of throwing an exception.
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,358 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.filesystem.PartitionTimeExtractor;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Supplier;
+
+import static org.apache.flink.table.filesystem.TimestampPartTimeExtractor.toLocalDateTime;
+
+/**
+ * This is the single (non-parallel) monitoring task which takes a {@link HiveTableInputFormat},
+ * it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring partitions of hive meta store.</li>
+ * <li>Deciding which partitions should be further read and processed.</li>
+ * <li>Creating the {@link HiveTableInputSplit splits} corresponding to those partitions.</li>
+ * <li>Assigning them to downstream tasks for further processing.</li>
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ContinuousFileReaderOperator}
+ * which can have parallelism greater than one.
+ *
+ * <p><b>IMPORTANT NOTE: </b> Splits are forwarded downstream for reading in ascending partition time order,
+ * based on the partition time of the partitions they belong to.
+ */
+public class HiveContinuousMonitoringFunction
+ extends RichSourceFunction<TimestampedHiveInputSplit>
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveContinuousMonitoringFunction.class);
+
+ /** The parallelism of the downstream readers. */
+ private final int readerParallelism;
+
+ /** The interval between consecutive path scans. */
+ private final long interval;
+
+ private final HiveShim hiveShim;
+
+ private final JobConfWrapper conf;
+
+ private final ObjectPath tablePath;
+
+ private final List<String> partitionKeys;
+
+ private final String[] fieldNames;
+
+ private final DataType[] fieldTypes;
+
+ // consumer variables
+ private final String consumeOrder;
+ private final String consumeOffset;
+
+ // extractor variables
+ private final String extractorType;
+ private final String extractorClass;
+ private final String extractorPattern;
+
+ private volatile boolean isRunning = true;
+
+ /** The maximum partition read time seen so far. */
+ private volatile long currentReadTime;
+
+ private transient PartitionDiscovery.Context context;
+
+ private transient PartitionDiscovery fetcher;
+
+ private transient Object checkpointLock;
+
+ private transient ListState<Long> currReadTimeState;
+
+ private transient ListState<List<List<String>>> distinctPartsState;
+
+ private transient IMetaStoreClient client;
+
+ private transient Properties tableProps;
+
+ private transient String defaultPartitionName;
+
+ private transient Set<List<String>> distinctPartitions;
+
+ public HiveContinuousMonitoringFunction(
+ HiveShim hiveShim,
+ JobConf conf,
+ ObjectPath tablePath,
+ CatalogTable catalogTable,
+ int readerParallelism,
+ String consumeOrder,
+ String consumeOffset,
+ String extractorType,
+ String extractorClass,
+ String extractorPattern,
+ long interval) {
+ this.hiveShim = hiveShim;
+ this.conf = new JobConfWrapper(conf);
+ this.tablePath = tablePath;
+ this.partitionKeys = catalogTable.getPartitionKeys();
+ this.fieldNames = catalogTable.getSchema().getFieldNames();
+ this.fieldTypes = catalogTable.getSchema().getFieldDataTypes();
+ this.consumeOrder = consumeOrder;
+ this.extractorType = extractorType;
+ this.extractorClass = extractorClass;
+ this.extractorPattern = extractorPattern;
+ this.consumeOffset = consumeOffset;
+
+ this.interval = interval;
+ this.readerParallelism = Math.max(readerParallelism, 1);
+ this.currentReadTime = 0;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.currReadTimeState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ LongSerializer.INSTANCE
+ )
+ );
+ this.distinctPartsState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
+ )
+ );
+
+ this.client = this.hiveShim.getHiveMetastoreClient(new HiveConf(conf.conf(), HiveConf.class));
+
+ Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
+ this.tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, hiveTable);
+ this.defaultPartitionName = conf.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+ HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
+
+ PartitionTimeExtractor extractor = PartitionTimeExtractor.create(
+ getRuntimeContext().getUserCodeClassLoader(),
+ extractorType,
+ extractorClass,
+ extractorPattern);
+
+ this.fetcher = new DirectoryMonitorDiscovery();
+
+ Path location = new Path(hiveTable.getSd().getLocation());
+ FileSystem fs = location.getFileSystem(conf.conf());
+ this.context = new PartitionDiscovery.Context() {
+
+ @Override
+ public List<String> partitionKeys() {
+ return partitionKeys;
+ }
+
+ @Override
+ public Partition getPartition(List<String> partValues) throws TException {
+ return client.getPartition(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ partValues);
+ }
+
+ @Override
+ public FileSystem fileSystem() {
+ return fs;
+ }
+
+ @Override
+ public Path tableLocation() {
+ return new Path(hiveTable.getSd().getLocation());
+ }
+
+ @Override
+ public long extractTimestamp(
+ List<String> partKeys,
+ List<String> partValues,
+ Supplier<Long> fileTime) {
+ switch (consumeOrder) {
+ case "create-time":
Review comment:
Should add constants for `create-time` and `partition-time`.
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,358 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.filesystem.PartitionTimeExtractor;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.function.Supplier;
+
+import static org.apache.flink.table.filesystem.TimestampPartTimeExtractor.toLocalDateTime;
+
+/**
+ * This is the single (non-parallel) monitoring task which takes a {@link HiveTableInputFormat},
+ * it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring partitions of hive meta store.</li>
+ * <li>Deciding which partitions should be further read and processed.</li>
+ * <li>Creating the {@link HiveTableInputSplit splits} corresponding to those partitions.</li>
+ * <li>Assigning them to downstream tasks for further processing.</li>
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ContinuousFileReaderOperator}
+ * which can have parallelism greater than one.
+ *
+ * <p><b>IMPORTANT NOTE: </b> Splits are forwarded downstream for reading in ascending partition time order,
+ * based on the partition time of the partitions they belong to.
+ */
+public class HiveContinuousMonitoringFunction
+ extends RichSourceFunction<TimestampedHiveInputSplit>
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveContinuousMonitoringFunction.class);
+
+ /** The parallelism of the downstream readers. */
+ private final int readerParallelism;
+
+ /** The interval between consecutive path scans. */
+ private final long interval;
+
+ private final HiveShim hiveShim;
+
+ private final JobConfWrapper conf;
+
+ private final ObjectPath tablePath;
+
+ private final List<String> partitionKeys;
+
+ private final String[] fieldNames;
+
+ private final DataType[] fieldTypes;
+
+ // consumer variables
+ private final String consumeOrder;
+ private final String consumeOffset;
+
+ // extractor variables
+ private final String extractorType;
+ private final String extractorClass;
+ private final String extractorPattern;
+
+ private volatile boolean isRunning = true;
+
+ /** The maximum partition read time seen so far. */
+ private volatile long currentReadTime;
+
+ private transient PartitionDiscovery.Context context;
+
+ private transient PartitionDiscovery fetcher;
+
+ private transient Object checkpointLock;
+
+ private transient ListState<Long> currReadTimeState;
+
+ private transient ListState<List<List<String>>> distinctPartsState;
+
+ private transient IMetaStoreClient client;
+
+ private transient Properties tableProps;
+
+ private transient String defaultPartitionName;
+
+ private transient Set<List<String>> distinctPartitions;
+
+ public HiveContinuousMonitoringFunction(
+ HiveShim hiveShim,
+ JobConf conf,
+ ObjectPath tablePath,
+ CatalogTable catalogTable,
+ int readerParallelism,
+ String consumeOrder,
+ String consumeOffset,
+ String extractorType,
+ String extractorClass,
+ String extractorPattern,
+ long interval) {
+ this.hiveShim = hiveShim;
+ this.conf = new JobConfWrapper(conf);
+ this.tablePath = tablePath;
+ this.partitionKeys = catalogTable.getPartitionKeys();
+ this.fieldNames = catalogTable.getSchema().getFieldNames();
+ this.fieldTypes = catalogTable.getSchema().getFieldDataTypes();
+ this.consumeOrder = consumeOrder;
+ this.extractorType = extractorType;
+ this.extractorClass = extractorClass;
+ this.extractorPattern = extractorPattern;
+ this.consumeOffset = consumeOffset;
+
+ this.interval = interval;
+ this.readerParallelism = Math.max(readerParallelism, 1);
+ this.currentReadTime = 0;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.currReadTimeState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ LongSerializer.INSTANCE
+ )
+ );
+ this.distinctPartsState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
+ )
+ );
+
+ this.client = this.hiveShim.getHiveMetastoreClient(new HiveConf(conf.conf(), HiveConf.class));
+
+ Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
+ this.tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, hiveTable);
+ this.defaultPartitionName = conf.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+ HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
+
+ PartitionTimeExtractor extractor = PartitionTimeExtractor.create(
+ getRuntimeContext().getUserCodeClassLoader(),
+ extractorType,
+ extractorClass,
+ extractorPattern);
+
+ this.fetcher = new DirectoryMonitorDiscovery();
+
+ Path location = new Path(hiveTable.getSd().getLocation());
+ FileSystem fs = location.getFileSystem(conf.conf());
+ this.context = new PartitionDiscovery.Context() {
+
+ @Override
+ public List<String> partitionKeys() {
+ return partitionKeys;
+ }
+
+ @Override
+ public Partition getPartition(List<String> partValues) throws TException {
+ return client.getPartition(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ partValues);
+ }
+
+ @Override
+ public FileSystem fileSystem() {
+ return fs;
+ }
+
+ @Override
+ public Path tableLocation() {
+ return new Path(hiveTable.getSd().getLocation());
+ }
+
+ @Override
+ public long extractTimestamp(
+ List<String> partKeys,
+ List<String> partValues,
+ Supplier<Long> fileTime) {
+ switch (consumeOrder) {
+ case "create-time":
+ return fileTime.get();
+ case "partition-time":
+ return toMills(extractor.extract(partKeys, partValues));
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported consumer order: " + consumeOrder);
+ }
+ }
+ };
+
+ this.distinctPartitions = new HashSet<>();
+ if (context.isRestored()) {
+ LOG.info("Restoring state for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.currReadTimeState.get().iterator().next();
+ this.distinctPartitions.addAll(this.distinctPartsState.get().iterator().next());
+ } else {
+ LOG.info("No state to restore for the {}.", getClass().getSimpleName());
+ if (consumeOffset != null) {
+ this.currentReadTime = toMills(toLocalDateTime(consumeOffset));
+ }
+ }
+ }
+
+ @Override
+ public void run(SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ checkpointLock = context.getCheckpointLock();
+ while (isRunning) {
+ synchronized (checkpointLock) {
+ monitorAndForwardSplits(context);
+ }
+ Thread.sleep(interval);
+ }
+ }
+
+ private void monitorAndForwardSplits(
+ SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ assert (Thread.holdsLock(checkpointLock));
+
+ List<Tuple2<Partition, Long>> partitions = fetcher.fetchPartitions(this.context, currentReadTime);
+
+ if (partitions.isEmpty()) {
+ return;
+ }
+
+ partitions.sort((o1, o2) -> (int) (o1.f1 - o2.f1));
+
+ long maxTimestamp = Long.MIN_VALUE;
+ Set<List<String>> nextDistinctParts = new HashSet<>();
+ for (Tuple2<Partition, Long> tuple2 : partitions) {
+ Partition partition = tuple2.f0;
+ List<String> partSpec = partition.getValues();
+ if (!this.distinctPartitions.contains(partSpec)) {
+ this.distinctPartitions.add(partition.getValues());
Review comment:
this.distinctPartitions.add(partSpec)?
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,280 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.connectors.hive.read.PartitionStrategy.PartitionStrategyFactory;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * This is the single (non-parallel) monitoring task which takes a {@link HiveTableInputFormat},
+ * it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring partitions of hive meta store.</li>
+ * <li>Deciding which partitions should be further read and processed.</li>
+ * <li>Creating the {@link HiveTableInputSplit splits} corresponding to those partitions.</li>
+ * <li>Assigning them to downstream tasks for further processing.</li>
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ContinuousFileReaderOperator}
+ * which can have parallelism greater than one.
+ *
+ * <p><b>IMPORTANT NOTE: </b> Splits are forwarded downstream for reading in ascending partition time order,
+ * based on the partition time of the partitions they belong to.
+ */
+public class HiveContinuousMonitoringFunction
+ extends RichSourceFunction<TimestampedHiveInputSplit>
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveContinuousMonitoringFunction.class);
+
+ /** The parallelism of the downstream readers. */
+ private final int readerParallelism;
+
+ /** The interval between consecutive path scans. */
+ private final long interval;
+
+ private final HiveShim hiveShim;
+
+ private final JobConfWrapper conf;
+
+ private final ObjectPath tablePath;
+
+ private final List<String> partitionKeys;
+
+ private final String[] fieldNames;
+
+ private final DataType[] fieldTypes;
+
+ private final long startupTimestampMillis;
+
+ private final PartitionStrategyFactory partStrategyFactory;
+
+ private volatile boolean isRunning = true;
+
+ /** The maximum partition read time seen so far. */
+ private volatile long currentReadTime;
+
+ private transient PartitionStrategy strategy;
+
+ private transient Object checkpointLock;
+
+ private transient ListState<Long> currReadTimeState;
+
+ private transient ListState<List<List<String>>> distinctPartsState;
+
+ private transient IMetaStoreClient client;
+
+ private transient Properties tableProps;
+
+ private transient String defaultPartitionName;
+
+ private transient Set<List<String>> distinctPartitions;
+
+ public HiveContinuousMonitoringFunction(
+ HiveShim hiveShim,
+ JobConf conf,
+ ObjectPath tablePath,
+ CatalogTable catalogTable,
+ long startupTimestampMillis,
+ PartitionStrategyFactory partStrategyFactory,
+ int readerParallelism,
+ long interval) {
+ this.hiveShim = hiveShim;
+ this.conf = new JobConfWrapper(conf);
+ this.tablePath = tablePath;
+ this.partitionKeys = catalogTable.getPartitionKeys();
+ this.fieldNames = catalogTable.getSchema().getFieldNames();
+ this.fieldTypes = catalogTable.getSchema().getFieldDataTypes();
+ this.startupTimestampMillis = startupTimestampMillis;
+ this.partStrategyFactory = partStrategyFactory;
+
+ this.interval = interval;
+ this.readerParallelism = Math.max(readerParallelism, 1);
+ this.currentReadTime = Long.MIN_VALUE;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.strategy = partStrategyFactory.createStrategy(getRuntimeContext().getUserCodeClassLoader());
+
+ this.currReadTimeState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ LongSerializer.INSTANCE
+ )
+ );
+ this.distinctPartsState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
+ )
+ );
+
+ this.client = this.hiveShim.getHiveMetastoreClient(new HiveConf(conf.conf(), HiveConf.class));
+
+ Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
+ this.tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, hiveTable);
+ this.defaultPartitionName = conf.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+ HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
+
+ this.distinctPartitions = new HashSet<>();
+ if (context.isRestored()) {
+ LOG.info("Restoring state for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.currReadTimeState.get().iterator().next();
+ this.distinctPartitions.addAll(this.distinctPartsState.get().iterator().next());
+ } else {
+ LOG.info("No state to restore for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.startupTimestampMillis;
+ }
+ }
+
+ @Override
+ public void run(SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ checkpointLock = context.getCheckpointLock();
+ while (isRunning) {
+ synchronized (checkpointLock) {
+ monitorAndForwardSplits(context);
+ }
+ Thread.sleep(interval);
+ }
+ }
+
+ private void monitorAndForwardSplits(
+ SourceContext<TimestampedHiveInputSplit> context) throws IOException, TException {
+ assert (Thread.holdsLock(checkpointLock));
+
+ List<Partition> partitions = client.listPartitionsByFilter(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ strategy.generateFetchFilter(partitionKeys, currentReadTime),
+ (short) -1);
+
+ if (partitions.isEmpty()) {
+ return;
+ }
+
+ long maxTime = Long.MIN_VALUE;
+ for (Partition partition : partitions) {
+ List<String> partSpec = partition.getValues();
+ if (!this.distinctPartitions.contains(partSpec)) {
+ this.distinctPartitions.add(partition.getValues());
+ long time = this.strategy.extractPartTime(partitionKeys, partition.getValues());
+ if (time > maxTime) {
+ maxTime = time;
+ }
+ HiveTableInputSplit[] splits = HiveTableInputFormat.createInputSplits(
+ this.readerParallelism,
+ Collections.singletonList(toHiveTablePartition(partition)),
+ this.conf.conf());
+ for (HiveTableInputSplit split : splits) {
+ context.collect(new TimestampedHiveInputSplit(time, split));
+ }
+ }
+ }
+ this.currentReadTime = maxTime;
+
+ this.distinctPartitions.removeIf(partSpec -> this.strategy.canExpireForDistinct(
+ this.strategy.extractPartTime(partitionKeys, partSpec),
+ this.currentReadTime));
+ }
+
+ private HiveTablePartition toHiveTablePartition(Partition p) {
+ return HiveTableSource.toHiveTablePartition(
+ partitionKeys, fieldNames, fieldTypes, hiveShim, tableProps, defaultPartitionName, p);
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ Preconditions.checkState(this.currReadTimeState != null,
+ "The " + getClass().getSimpleName() + " state has not been properly initialized.");
+
+ this.currReadTimeState.clear();
+ this.currReadTimeState.add(this.currentReadTime);
+
+ this.distinctPartsState.clear();
+ this.distinctPartsState.add(new ArrayList<>(this.distinctPartitions));
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("{} checkpointed {}.", getClass().getSimpleName(), currentReadTime);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+
+ if (checkpointLock != null) {
+ synchronized (checkpointLock) {
+ currentReadTime = Long.MAX_VALUE;
+ isRunning = false;
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ if (checkpointLock != null) {
+ // this is to cover the case where cancel() is called before the run()
Review comment:
Let's move this comment to the outside of the if block, so that it's easier to understand.
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/SplitReader.java
##########
@@ -46,4 +46,17 @@
* @throws IOException Thrown, if an I/O error occurred.
*/
RowData nextRecord(RowData reuse) throws IOException;
+
+ /**
+ * Seek to a particular row number.
+ */
+ default void seekToRow(long rowCount, RowData reuse) throws IOException {
+ for (int i = 0; i < rowCount; i++) {
+ boolean end = reachedEnd();
+ if (end) {
+ throw new RuntimeException("Seek to many rows.");
Review comment:
"Seek too many rows." ?
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/TimestampedHiveInputSplit.java
##########
@@ -0,0 +1,232 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.streaming.api.functions.source.TimestampedInputSplit;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.lang.reflect.Type;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * A {@link HiveTableInputSplit} with {@link TimestampedInputSplit}.
+ * Kryo serializer can not deal with hadoop split, need specific type information factory.
+ */
+@TypeInfo(TimestampedHiveInputSplit.SplitTypeInfoFactory.class)
+public class TimestampedHiveInputSplit extends HiveTableInputSplit implements TimestampedInputSplit {
+
+ private static final long serialVersionUID = 1L;
+
+ /** The modification time of the file this split belongs to. */
+ private final long modificationTime;
+
+ /**
+ * The state of the split. This information is used when
+ * restoring from a checkpoint and allows to resume reading the
+ * underlying file from the point we left off.
+ * */
+ private Serializable splitState;
+
+ public TimestampedHiveInputSplit(
+ long modificationTime,
+ HiveTableInputSplit split) {
+ super(
+ split.getSplitNumber(),
+ split.getHadoopInputSplit(),
+ split.getJobConf(),
+ split.getHiveTablePartition());
+ this.modificationTime = modificationTime;
+ }
+
+ @Override
+ public void setSplitState(Serializable state) {
+ this.splitState = state;
+ }
+
+ @Override
+ public Serializable getSplitState() {
+ return this.splitState;
+ }
+
+ @Override
+ public long getModificationTime() {
+ return modificationTime;
+ }
+
+ @Override
+ public int compareTo(TimestampedInputSplit o) {
+ TimestampedHiveInputSplit split = (TimestampedHiveInputSplit) o;
+ int modTimeComp = Long.compare(this.modificationTime, split.modificationTime);
+ if (modTimeComp != 0L) {
+ return modTimeComp;
+ }
+
+ int pathComp = this.hiveTablePartition.getStorageDescriptor().compareTo(
+ split.hiveTablePartition.getStorageDescriptor());
+
+ return pathComp != 0 ? pathComp :
+ this.getSplitNumber() - o.getSplitNumber();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
Review comment:
For two split instances, this means `equals` can return false while `compareTo` returns 0. Is this intended behavior?
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/TimestampedHiveInputSplit.java
##########
@@ -0,0 +1,232 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.SimpleTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.streaming.api.functions.source.TimestampedInputSplit;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.lang.reflect.Type;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * A {@link HiveTableInputSplit} with {@link TimestampedInputSplit}.
+ * Kryo serializer can not deal with hadoop split, need specific type information factory.
+ */
+@TypeInfo(TimestampedHiveInputSplit.SplitTypeInfoFactory.class)
+public class TimestampedHiveInputSplit extends HiveTableInputSplit implements TimestampedInputSplit {
+
+ private static final long serialVersionUID = 1L;
+
+ /** The modification time of the file this split belongs to. */
+ private final long modificationTime;
+
+ /**
+ * The state of the split. This information is used when
+ * restoring from a checkpoint and allows to resume reading the
+ * underlying file from the point we left off.
+ * */
+ private Serializable splitState;
+
+ public TimestampedHiveInputSplit(
+ long modificationTime,
+ HiveTableInputSplit split) {
+ super(
+ split.getSplitNumber(),
+ split.getHadoopInputSplit(),
+ split.getJobConf(),
+ split.getHiveTablePartition());
+ this.modificationTime = modificationTime;
+ }
+
+ @Override
+ public void setSplitState(Serializable state) {
+ this.splitState = state;
+ }
+
+ @Override
+ public Serializable getSplitState() {
+ return this.splitState;
+ }
+
+ @Override
+ public long getModificationTime() {
+ return modificationTime;
+ }
+
+ @Override
+ public int compareTo(TimestampedInputSplit o) {
+ TimestampedHiveInputSplit split = (TimestampedHiveInputSplit) o;
+ int modTimeComp = Long.compare(this.modificationTime, split.modificationTime);
+ if (modTimeComp != 0L) {
+ return modTimeComp;
+ }
+
+ int pathComp = this.hiveTablePartition.getStorageDescriptor().compareTo(
Review comment:
I think this compares more than just paths.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.flink.table.filesystem;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import java.time.Duration;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds configuration constants used by filesystem(Including hive) connector.
+ */
+public class FileSystemOptions {
+
+ public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
+ key("streaming-source.enable")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Enable streaming source or not.");
+
+ public static final ConfigOption<Duration> STREAMING_SOURCE_MONITOR_INTERVAL =
+ key("streaming-source.monitor-interval")
+ .durationType()
+ .defaultValue(Duration.ofMillis(1))
+ .withDescription("The minimum interval allowed between consecutive partition/file discovery.");
Review comment:
Why it's the minimum interval? It may seem confusing to users.
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.flink.table.filesystem;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import java.time.Duration;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds configuration constants used by filesystem(Including hive) connector.
+ */
+public class FileSystemOptions {
+
+ public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
Review comment:
Wondering what's the relationship between this and the execution mode? E.g. can I enable streaming source with batch execution mode?
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.flink.table.filesystem;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import java.time.Duration;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds configuration constants used by filesystem(Including hive) connector.
+ */
+public class FileSystemOptions {
+
+ public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
+ key("streaming-source.enable")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Enable streaming source or not.");
+
+ public static final ConfigOption<Duration> STREAMING_SOURCE_MONITOR_INTERVAL =
+ key("streaming-source.monitor-interval")
+ .durationType()
+ .defaultValue(Duration.ofMillis(1))
+ .withDescription("The minimum interval allowed between consecutive partition/file discovery.");
+
+ public static final ConfigOption<String> STREAMING_SOURCE_CONSUME_ORDER =
+ key("streaming-source.consume-order")
+ .stringType()
+ .defaultValue("create-time")
+ .withDescription("The consume order of streaming source," +
+ " support create-time and partition-time." +
+ " create-time compare partition/file creation time;" +
+ " partition-time compare time represented by partition name.");
+
+ public static final ConfigOption<String> STREAMING_SOURCE_CONSUME_START_OFFSET =
+ key("streaming-source.consume-start-offset")
+ .stringType()
+ .defaultValue("1970-00-00")
+ .withDescription("Start offset for streaming consuming." +
+ " How to parse and compare offsets depends on your order." +
+ " For create-time and partition-time, should be a timestamp string.");
+
+ public static final ConfigOption<String> PARTITION_TIME_EXTRACTOR_TYPE =
+ key("partition.time-extractor.type")
+ .stringType()
+ .defaultValue("timestamp")
+ .withDescription("Time extractor to extract time from partition values." +
Review comment:
Clarify this should only be used if `STREAMING_SOURCE_CONSUME_ORDER` is set to `partition-time`?
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.flink.table.filesystem;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import java.time.Duration;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds configuration constants used by filesystem(Including hive) connector.
+ */
+public class FileSystemOptions {
+
+ public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
+ key("streaming-source.enable")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Enable streaming source or not.");
+
+ public static final ConfigOption<Duration> STREAMING_SOURCE_MONITOR_INTERVAL =
+ key("streaming-source.monitor-interval")
+ .durationType()
+ .defaultValue(Duration.ofMillis(1))
+ .withDescription("The minimum interval allowed between consecutive partition/file discovery.");
+
+ public static final ConfigOption<String> STREAMING_SOURCE_CONSUME_ORDER =
+ key("streaming-source.consume-order")
+ .stringType()
+ .defaultValue("create-time")
+ .withDescription("The consume order of streaming source," +
+ " support create-time and partition-time." +
+ " create-time compare partition/file creation time;" +
+ " partition-time compare time represented by partition name.");
+
+ public static final ConfigOption<String> STREAMING_SOURCE_CONSUME_START_OFFSET =
+ key("streaming-source.consume-start-offset")
+ .stringType()
+ .defaultValue("1970-00-00")
+ .withDescription("Start offset for streaming consuming." +
+ " How to parse and compare offsets depends on your order." +
+ " For create-time and partition-time, should be a timestamp string.");
+
+ public static final ConfigOption<String> PARTITION_TIME_EXTRACTOR_TYPE =
+ key("partition.time-extractor.type")
+ .stringType()
+ .defaultValue("timestamp")
+ .withDescription("Time extractor to extract time from partition values." +
+ " Support timestamp and custom." +
+ " For timestamp, can configure timestamp pattern." +
+ " For custom, should configure extractor class.");
+
+ public static final ConfigOption<String> PARTITION_TIME_EXTRACTOR_CLASS =
+ key("partition.time-extractor.class")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The extractor class for implement PartitionTimeExtractor interface.");
+
+ public static final ConfigOption<String> PARTITION_TIME_EXTRACTOR_TIMESTAMP_PATTERN =
+ key("partition.time-extractor.timestamp-pattern")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The 'timestamp' construction way allows users to use partition" +
+ " fields to get a legal timestamp pattern." +
+ " Default support 'yyyy-mm-dd hh:mm:ss' from first field." +
+ " If timestamp in partition is single field 'dt', can configure: '$dt'." +
+ " If timestamp in partition is year, month, day, hour," +
+ " can configure: '$year-$month-$day $hour:00:00'." +
+ " If timestamp in partition is dt and hour, can configure: '$dt $hour:00:00'.");
+
+ public static final ConfigOption<Duration> PARTITION_TIME_INTERVAL =
Review comment:
Seems unused
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.flink.table.filesystem;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import java.time.Duration;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds configuration constants used by filesystem(Including hive) connector.
+ */
+public class FileSystemOptions {
+
+ public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
+ key("streaming-source.enable")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Enable streaming source or not.");
+
+ public static final ConfigOption<Duration> STREAMING_SOURCE_MONITOR_INTERVAL =
+ key("streaming-source.monitor-interval")
+ .durationType()
+ .defaultValue(Duration.ofMillis(1))
+ .withDescription("The minimum interval allowed between consecutive partition/file discovery.");
+
+ public static final ConfigOption<String> STREAMING_SOURCE_CONSUME_ORDER =
+ key("streaming-source.consume-order")
+ .stringType()
+ .defaultValue("create-time")
+ .withDescription("The consume order of streaming source," +
+ " support create-time and partition-time." +
+ " create-time compare partition/file creation time;" +
+ " partition-time compare time represented by partition name.");
+
+ public static final ConfigOption<String> STREAMING_SOURCE_CONSUME_START_OFFSET =
+ key("streaming-source.consume-start-offset")
+ .stringType()
+ .defaultValue("1970-00-00")
+ .withDescription("Start offset for streaming consuming." +
+ " How to parse and compare offsets depends on your order." +
+ " For create-time and partition-time, should be a timestamp string.");
+
+ public static final ConfigOption<String> PARTITION_TIME_EXTRACTOR_TYPE =
+ key("partition.time-extractor.type")
+ .stringType()
+ .defaultValue("timestamp")
+ .withDescription("Time extractor to extract time from partition values." +
+ " Support timestamp and custom." +
Review comment:
I find `timestamp` somehow difficult to understand. According to the `PartitionTimeExtractor` interface, any implementation should extract a "timestamp". We may as well just call it `default`?
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/PartitionTimeExtractor.java
##########
@@ -0,0 +1,58 @@
+/*
+ * 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.flink.table.filesystem;
+
+import org.apache.flink.annotation.Experimental;
+
+import java.io.Serializable;
+import java.time.LocalDateTime;
+import java.util.List;
+
+/**
+ * Time extractor to extract time from partition values.
+ */
+@Experimental
+public interface PartitionTimeExtractor extends Serializable {
+
+ /**
+ * Extract time from partition keys and values.
+ */
+ LocalDateTime extract(List<String> partitionKeys, List<String> partitionValues);
+
+ static PartitionTimeExtractor create(
+ ClassLoader userClassLoader,
+ String extractorType,
+ String extractorClass,
+ String extractorPattern) {
+ switch (extractorType) {
+ case "timestamp":
Review comment:
Should add constants for these values
##########
File path: flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.flink.table.filesystem;
+
+import org.apache.flink.configuration.ConfigOption;
+
+import java.time.Duration;
+
+import static org.apache.flink.configuration.ConfigOptions.key;
+
+/**
+ * This class holds configuration constants used by filesystem(Including hive) connector.
+ */
+public class FileSystemOptions {
+
+ public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
+ key("streaming-source.enable")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Enable streaming source or not.");
+
+ public static final ConfigOption<Duration> STREAMING_SOURCE_MONITOR_INTERVAL =
+ key("streaming-source.monitor-interval")
+ .durationType()
+ .defaultValue(Duration.ofMillis(1))
+ .withDescription("The minimum interval allowed between consecutive partition/file discovery.");
+
+ public static final ConfigOption<String> STREAMING_SOURCE_CONSUME_ORDER =
+ key("streaming-source.consume-order")
+ .stringType()
+ .defaultValue("create-time")
+ .withDescription("The consume order of streaming source," +
+ " support create-time and partition-time." +
+ " create-time compare partition/file creation time;" +
Review comment:
I think we'd better clarify this is not the partition create time in HMS, but the folder/file create time in filesystem?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 42953325bb1d6a447ef53e020d3cc311a8c986c6 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779)
* 44df52a0ff62e1b4ec65efd6a14a8b07d7712e75 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-626123587
Updated, @lirui-apache @openinx
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422748256
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveTableInputFormat.java
##########
@@ -248,7 +263,11 @@ public RowData nextRecord(RowData reuse) throws IOException {
}
@Override
- public void close() throws IOException {
+ public void close() {
Review comment:
I am wrong, should in close.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422454432
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/read/HiveContinuousMonitoringFunction.java
##########
@@ -0,0 +1,280 @@
+/*
+ * 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.flink.connectors.hive.read;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.connectors.hive.HiveTablePartition;
+import org.apache.flink.connectors.hive.HiveTableSource;
+import org.apache.flink.connectors.hive.JobConfWrapper;
+import org.apache.flink.connectors.hive.read.PartitionStrategy.PartitionStrategyFactory;
+import org.apache.flink.runtime.state.FunctionInitializationContext;
+import org.apache.flink.runtime.state.FunctionSnapshotContext;
+import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
+import org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator;
+import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.hive.client.HiveShim;
+import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * This is the single (non-parallel) monitoring task which takes a {@link HiveTableInputFormat},
+ * it is responsible for:
+ *
+ * <ol>
+ * <li>Monitoring partitions of hive meta store.</li>
+ * <li>Deciding which partitions should be further read and processed.</li>
+ * <li>Creating the {@link HiveTableInputSplit splits} corresponding to those partitions.</li>
+ * <li>Assigning them to downstream tasks for further processing.</li>
+ * </ol>
+ *
+ * <p>The splits to be read are forwarded to the downstream {@link ContinuousFileReaderOperator}
+ * which can have parallelism greater than one.
+ *
+ * <p><b>IMPORTANT NOTE: </b> Splits are forwarded downstream for reading in ascending partition time order,
+ * based on the partition time of the partitions they belong to.
+ */
+public class HiveContinuousMonitoringFunction
+ extends RichSourceFunction<TimestampedHiveInputSplit>
+ implements CheckpointedFunction {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveContinuousMonitoringFunction.class);
+
+ /** The parallelism of the downstream readers. */
+ private final int readerParallelism;
+
+ /** The interval between consecutive path scans. */
+ private final long interval;
+
+ private final HiveShim hiveShim;
+
+ private final JobConfWrapper conf;
+
+ private final ObjectPath tablePath;
+
+ private final List<String> partitionKeys;
+
+ private final String[] fieldNames;
+
+ private final DataType[] fieldTypes;
+
+ private final long startupTimestampMillis;
+
+ private final PartitionStrategyFactory partStrategyFactory;
+
+ private volatile boolean isRunning = true;
+
+ /** The maximum partition read time seen so far. */
+ private volatile long currentReadTime;
+
+ private transient PartitionStrategy strategy;
+
+ private transient Object checkpointLock;
+
+ private transient ListState<Long> currReadTimeState;
+
+ private transient ListState<List<List<String>>> distinctPartsState;
+
+ private transient IMetaStoreClient client;
+
+ private transient Properties tableProps;
+
+ private transient String defaultPartitionName;
+
+ private transient Set<List<String>> distinctPartitions;
+
+ public HiveContinuousMonitoringFunction(
+ HiveShim hiveShim,
+ JobConf conf,
+ ObjectPath tablePath,
+ CatalogTable catalogTable,
+ long startupTimestampMillis,
+ PartitionStrategyFactory partStrategyFactory,
+ int readerParallelism,
+ long interval) {
+ this.hiveShim = hiveShim;
+ this.conf = new JobConfWrapper(conf);
+ this.tablePath = tablePath;
+ this.partitionKeys = catalogTable.getPartitionKeys();
+ this.fieldNames = catalogTable.getSchema().getFieldNames();
+ this.fieldTypes = catalogTable.getSchema().getFieldDataTypes();
+ this.startupTimestampMillis = startupTimestampMillis;
+ this.partStrategyFactory = partStrategyFactory;
+
+ this.interval = interval;
+ this.readerParallelism = Math.max(readerParallelism, 1);
+ this.currentReadTime = Long.MIN_VALUE;
+ }
+
+ @Override
+ public void initializeState(FunctionInitializationContext context) throws Exception {
+ this.strategy = partStrategyFactory.createStrategy(getRuntimeContext().getUserCodeClassLoader());
+
+ this.currReadTimeState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ LongSerializer.INSTANCE
+ )
+ );
+ this.distinctPartsState = context.getOperatorStateStore().getListState(
+ new ListStateDescriptor<>(
+ "partition-monitoring-state",
+ new ListSerializer<>(new ListSerializer<>(StringSerializer.INSTANCE))
+ )
+ );
+
+ this.client = this.hiveShim.getHiveMetastoreClient(new HiveConf(conf.conf(), HiveConf.class));
+
+ Table hiveTable = client.getTable(tablePath.getDatabaseName(), tablePath.getObjectName());
+ this.tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, hiveTable);
+ this.defaultPartitionName = conf.conf().get(HiveConf.ConfVars.DEFAULTPARTITIONNAME.varname,
+ HiveConf.ConfVars.DEFAULTPARTITIONNAME.defaultStrVal);
+
+ this.distinctPartitions = new HashSet<>();
+ if (context.isRestored()) {
+ LOG.info("Restoring state for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.currReadTimeState.get().iterator().next();
+ this.distinctPartitions.addAll(this.distinctPartsState.get().iterator().next());
+ } else {
+ LOG.info("No state to restore for the {}.", getClass().getSimpleName());
+ this.currentReadTime = this.startupTimestampMillis;
+ }
+ }
+
+ @Override
+ public void run(SourceContext<TimestampedHiveInputSplit> context) throws Exception {
+ checkpointLock = context.getCheckpointLock();
+ while (isRunning) {
+ synchronized (checkpointLock) {
+ monitorAndForwardSplits(context);
+ }
+ Thread.sleep(interval);
+ }
+ }
+
+ private void monitorAndForwardSplits(
+ SourceContext<TimestampedHiveInputSplit> context) throws IOException, TException {
+ assert (Thread.holdsLock(checkpointLock));
+
+ List<Partition> partitions = client.listPartitionsByFilter(
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ strategy.generateFetchFilter(partitionKeys, currentReadTime),
+ (short) -1);
+
+ if (partitions.isEmpty()) {
+ return;
+ }
+
+ long maxTime = Long.MIN_VALUE;
+ for (Partition partition : partitions) {
+ List<String> partSpec = partition.getValues();
+ if (!this.distinctPartitions.contains(partSpec)) {
+ this.distinctPartitions.add(partition.getValues());
+ long time = this.strategy.extractPartTime(partitionKeys, partition.getValues());
+ if (time > maxTime) {
+ maxTime = time;
+ }
+ HiveTableInputSplit[] splits = HiveTableInputFormat.createInputSplits(
+ this.readerParallelism,
+ Collections.singletonList(toHiveTablePartition(partition)),
+ this.conf.conf());
+ for (HiveTableInputSplit split : splits) {
+ context.collect(new TimestampedHiveInputSplit(time, split));
+ }
+ }
+ }
+ this.currentReadTime = maxTime;
+
+ this.distinctPartitions.removeIf(partSpec -> this.strategy.canExpireForDistinct(
+ this.strategy.extractPartTime(partitionKeys, partSpec),
+ this.currentReadTime));
+ }
+
+ private HiveTablePartition toHiveTablePartition(Partition p) {
+ return HiveTableSource.toHiveTablePartition(
+ partitionKeys, fieldNames, fieldTypes, hiveShim, tableProps, defaultPartitionName, p);
+ }
+
+ @Override
+ public void snapshotState(FunctionSnapshotContext context) throws Exception {
+ Preconditions.checkState(this.currReadTimeState != null,
+ "The " + getClass().getSimpleName() + " state has not been properly initialized.");
+
+ this.currReadTimeState.clear();
+ this.currReadTimeState.add(this.currentReadTime);
+
+ this.distinctPartsState.clear();
+ this.distinctPartsState.add(new ArrayList<>(this.distinctPartitions));
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("{} checkpointed {}.", getClass().getSimpleName(), currentReadTime);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+
+ if (checkpointLock != null) {
+ synchronized (checkpointLock) {
+ currentReadTime = Long.MAX_VALUE;
+ isRunning = false;
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ if (checkpointLock != null) {
+ // this is to cover the case where cancel() is called before the run()
Review comment:
this is from another thread.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #12004:
URL: https://github.com/apache/flink/pull/12004#discussion_r422462727
##########
File path: flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSource.java
##########
@@ -181,6 +214,47 @@ public boolean isBounded() {
return source.name(explainSource());
}
+ private DataStream<RowData> createStreamSource(
+ StreamExecutionEnvironment execEnv,
+ TypeInformation<RowData> typeInfo,
+ HiveTableInputFormat inputFormat) {
+ final Map<String, String> properties = catalogTable.getProperties();
+ PartitionFetcherFactory strategyFactory = cl -> PartitionFetcher.createStrategy(
+ properties.get(HIVE_STREAMING_SOURCE_PARTITION_STRATEGY.key()),
+ properties.get(HIVE_STREAMING_SOURCE_PARTITION_STRATEGY_CLASS.key()),
+ cl);
+
+ String monitorIntervalStr = properties.get(HIVE_STREAMING_SOURCE_MONITOR_INTERVAL.key());
+ Duration monitorInterval = monitorIntervalStr != null ?
+ TimeUtils.parseDuration(monitorIntervalStr) :
+ HIVE_STREAMING_SOURCE_MONITOR_INTERVAL.defaultValue();
+
+ HiveContinuousMonitoringFunction monitoringFunction = new HiveContinuousMonitoringFunction(
+ hiveShim,
+ jobConf,
+ tablePath,
+ catalogTable,
+ getStartupPartition(),
+ strategyFactory,
+ execEnv.getParallelism(),
+ monitorInterval.toMillis());
+
+ ContinuousFileReaderOperatorFactory<RowData, TimestampedHiveInputSplit> factory =
+ new ContinuousFileReaderOperatorFactory<>(inputFormat);
Review comment:
> we don't do any split inside the file
We will split one file to many splits.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 42953325bb1d6a447ef53e020d3cc311a8c986c6 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779)
* 44df52a0ff62e1b4ec65efd6a14a8b07d7712e75 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
}, {
"hash" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"triggerType" : "PUSH"
}, {
"hash" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929",
"triggerID" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"triggerType" : "PUSH"
}, {
"hash" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937",
"triggerID" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"triggerType" : "PUSH"
}, {
"hash" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=946",
"triggerID" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"triggerType" : "PUSH"
}, {
"hash" : "bffea7d6cacc503cc630c8b3ddf77f4093554e84",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=967",
"triggerID" : "bffea7d6cacc503cc630c8b3ddf77f4093554e84",
"triggerType" : "PUSH"
}, {
"hash" : "48017434485d6059dafddc8617bf7f75fa7b3892",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=982",
"triggerID" : "48017434485d6059dafddc8617bf7f75fa7b3892",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3ac79a41395792dfecefc87a19c58468e91ede8d UNKNOWN
* 006610cb59c3665b510485c6d6ad883d91fc3e6d Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=946)
* bffea7d6cacc503cc630c8b3ddf77f4093554e84 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=967)
* 48017434485d6059dafddc8617bf7f75fa7b3892 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=982)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4223a26640c7243a61135e803aa1dd21888114a5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683)
* 42953325bb1d6a447ef53e020d3cc311a8c986c6 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a9a2c659b1693752361aaf3cf6f67e8251032983 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669)
* 4223a26640c7243a61135e803aa1dd21888114a5 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a9a2c659b1693752361aaf3cf6f67e8251032983 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669)
* 4223a26640c7243a61135e803aa1dd21888114a5 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #12004: [FLINK-17434][core][hive] Hive partitioned source support streaming read
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #12004:
URL: https://github.com/apache/flink/pull/12004#issuecomment-624445192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=666",
"triggerID" : "dfa5dc1b96f0e1a13b9c31b4fab97cf58d580299",
"triggerType" : "PUSH"
}, {
"hash" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=669",
"triggerID" : "a9a2c659b1693752361aaf3cf6f67e8251032983",
"triggerType" : "PUSH"
}, {
"hash" : "4223a26640c7243a61135e803aa1dd21888114a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=683",
"triggerID" : "4223a26640c7243a61135e803aa1dd21888114a5",
"triggerType" : "PUSH"
}, {
"hash" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=779",
"triggerID" : "42953325bb1d6a447ef53e020d3cc311a8c986c6",
"triggerType" : "PUSH"
}, {
"hash" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=872",
"triggerID" : "44df52a0ff62e1b4ec65efd6a14a8b07d7712e75",
"triggerType" : "PUSH"
}, {
"hash" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "3ac79a41395792dfecefc87a19c58468e91ede8d",
"triggerType" : "PUSH"
}, {
"hash" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=929",
"triggerID" : "b3802f894dcdd7226316583e66b37a662c3e5bdb",
"triggerType" : "PUSH"
}, {
"hash" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937",
"triggerID" : "6545efc0050c0d2ae84755199cf3b5fd55e82dca",
"triggerType" : "PUSH"
}, {
"hash" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=946",
"triggerID" : "006610cb59c3665b510485c6d6ad883d91fc3e6d",
"triggerType" : "PUSH"
}, {
"hash" : "bffea7d6cacc503cc630c8b3ddf77f4093554e84",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "bffea7d6cacc503cc630c8b3ddf77f4093554e84",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 3ac79a41395792dfecefc87a19c58468e91ede8d UNKNOWN
* 6545efc0050c0d2ae84755199cf3b5fd55e82dca Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=937)
* 006610cb59c3665b510485c6d6ad883d91fc3e6d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=946)
* bffea7d6cacc503cc630c8b3ddf77f4093554e84 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org