[jira] [Updated] (FLINK-20953) canal-json.table.include Supports regular or wildcard characters
[ https://issues.apache.org/jira/browse/FLINK-20953?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] wangfei updated FLINK-20953: Description: |h5. canal-json.database.include|optional|(none)|String|Only read changelog rows which match the specific database (by comparing the "database" meta field in the Canal record).| |h5. canal-json.table.include|optional|(none)|String|Only read changelog rows which match the specific table (by comparing the "table" meta field in the Canal record).| h5. canal-json.database.include and canal-json.table.include Supports regular or wildcard characters was: |h5. canal-json.database.include|optional|(none)|String|仅读取指定数据库的 changelog 记录（通过对比 Canal 记录中的 "database" 元数据字段）| |h5. canal-json.table.include|optional|(none)|String|仅读取指定表的 changelog 记录（通过对比 Canal 记录中的 "table" 元数据字段）。| h5. canal-json.database.include and canal-json.table.include Supports regular or wildcard characters > canal-json.table.include Supports regular or wildcard characters > > > Key: FLINK-20953 > URL: https://issues.apache.org/jira/browse/FLINK-20953 > Project: Flink > Issue Type: Improvement > Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile) >Reporter: wangfei >Priority: Major > Fix For: 1.13.0 > > > |h5. canal-json.database.include|optional|(none)|String|Only read changelog > rows which match the specific database (by comparing the "database" meta > field in the Canal record).| > |h5. canal-json.table.include|optional|(none)|String|Only read changelog rows > which match the specific table (by comparing the "table" meta field in the > Canal record).| > h5. canal-json.database.include and canal-json.table.include Supports > regular or wildcard characters > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] wuchong commented on a change in pull request #14536: [FLINK-20812][Connector][Hbase] hbase in sql mode,can use 'properties.*' add Configuration parameter.
wuchong commented on a change in pull request #14536:
URL: https://github.com/apache/flink/pull/14536#discussion_r556316339
##
File path:
flink-connectors/flink-connector-hbase-base/src/main/java/org/apache/flink/connector/hbase/source/AbstractHBaseDynamicTableSource.java
##
@@ -114,4 +117,28 @@ public String asSummaryString() {
public HBaseTableSchema getHBaseTableSchema() {
return this.hbaseSchema;
}
+
+// get HBase table properties which start with prefix
+public static Properties getHBaseClientProperties(
+Map tableOptions, String prefix) {
Review comment:
I think we don't need the `prefix` parameter, the prefix should always
be `properties.`.
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:
us...@infra.apache.org
[jira] [Commented] (FLINK-20953) canal-json.table.include Supports regular or wildcard characters
[ https://issues.apache.org/jira/browse/FLINK-20953?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263963#comment-17263963 ] wangfei commented on FLINK-20953: - @[~jark] @[~nicholasjiang] > canal-json.table.include Supports regular or wildcard characters > > > Key: FLINK-20953 > URL: https://issues.apache.org/jira/browse/FLINK-20953 > Project: Flink > Issue Type: Improvement > Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile) >Reporter: wangfei >Priority: Major > Fix For: 1.13.0 > > > |h5. canal-json.database.include|optional|(none)|String|仅读取指定数据库的 changelog > 记录（通过对比 Canal 记录中的 "database" 元数据字段）| > |h5. canal-json.table.include|optional|(none)|String|仅读取指定表的 changelog > 记录（通过对比 Canal 记录中的 "table" 元数据字段）。| > h5. canal-json.database.include and canal-json.table.include Supports > regular or wildcard characters > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot commented on pull request #14627: [FLINK-20946][python] Optimize Python ValueState Implementation In PyFlink
flinkbot commented on pull request #14627: URL: https://github.com/apache/flink/pull/14627#issuecomment-759272918 ## CI report: * a0e6a39605a3d1a80f75f8764534dfb82d08c31f UNKNOWN Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #14626: [FLINK-20948][table-planner-blink] Introduce StreamPhysicalDeduplicate, and make StreamExecDeduplicate only extended from ExecNode
flinkbot commented on pull request #14626: URL: https://github.com/apache/flink/pull/14626#issuecomment-759272820 ## CI report: * 1387747f865a79d120ee7f5d6f24685c90e0076f UNKNOWN Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] SteNicholas commented on pull request #14615: [FLINK-20930][runtime] Remove AbstractExecutionSlotAllocator
SteNicholas commented on pull request #14615: URL: https://github.com/apache/flink/pull/14615#issuecomment-759272794 @tillrohrmann , could you please help to review this simple pull request? 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14620: [FLINK-20940][table-planner] Use session time zone in LOCALTIME/LOCALTIMSTAMP functions
flinkbot edited a comment on pull request #14620: URL: https://github.com/apache/flink/pull/14620#issuecomment-758696085 ## CI report: * 9476884989f58f1f3ecc2bf33e9d225524d3d0b1 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11941) * a23b91048435a844d5c7b8acb174a3424c78a68a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11961) * ed1cd707bd4e584d02433819a05470beb7a7e64c UNKNOWN Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14616: [FLINK-20879][table-api] Use MemorySize type instead of String type for memory ConfigOption in ExecutionConfigOptions
flinkbot edited a comment on pull request #14616: URL: https://github.com/apache/flink/pull/14616#issuecomment-758526377 ## CI report: * 448c026a402e045e050f405daf934a8a7c880c9d UNKNOWN * 088770300dee0faec4b956833dccef115b05bbb7 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11955) * 37e1dac970b011ef214d7fee59b36d7f4646b0e1 UNKNOWN Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14606: [Flink-20876][table-planner-blink] Separate the implementation of StreamExecTemporalJoin
flinkbot edited a comment on pull request #14606: URL: https://github.com/apache/flink/pull/14606#issuecomment-757754275 ## CI report: * 3776b52cfe3535dcc193b3a922a7d1d658126d66 UNKNOWN * 155b18c169e45a97cd52c5b43883d5cf6b79f038 UNKNOWN * 9276dee92aa80400b79a80168d19553aa2689a10 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11964) Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14604: [FLINK-20861][format/json] Provide an option for serializing DECIMALs in JSON as plain number instead of scientific notation
flinkbot edited a comment on pull request #14604: URL: https://github.com/apache/flink/pull/14604#issuecomment-757521857 ## CI report: * 6f16ecba1741139347662f1a892cca2885b603c9 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11862) * 34a633e801bcb8100e0962b358b4e400a47c7047 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11959) * 095aa26dba7a04800cb203e252774d0d7655cc25 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11962) * 14ecbcb1bc4fc86bcdda844a48b8c187f6c9a96a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11968) Bot commands 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 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: us...@infra.apache.org
[jira] [Created] (FLINK-20953) canal-json.table.include Supports regular or wildcard characters
wangfei created FLINK-20953: --- Summary: canal-json.table.include Supports regular or wildcard characters Key: FLINK-20953 URL: https://issues.apache.org/jira/browse/FLINK-20953 Project: Flink Issue Type: Improvement Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile) Reporter: wangfei Fix For: 1.13.0 |h5. canal-json.database.include|optional|(none)|String|仅读取指定数据库的 changelog 记录（通过对比 Canal 记录中的 "database" 元数据字段）| |h5. canal-json.table.include|optional|(none)|String|仅读取指定表的 changelog 记录（通过对比 Canal 记录中的 "table" 元数据字段）。| h5. canal-json.database.include and canal-json.table.include Supports regular or wildcard characters -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] XComp commented on a change in pull request #14584: [FLINK-20850][runtime] Removing usage of CoLocationConstraints
XComp commented on a change in pull request #14584:
URL: https://github.com/apache/flink/pull/14584#discussion_r556320209
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
##
@@ -1018,10 +1018,10 @@ private void setCoLocation() {
"Cannot use a co-location constraint without a
slot sharing group");
}
-Tuple2 constraint =
+Tuple2 constraint =
coLocationGroups.computeIfAbsent(
coLocationGroupKey,
-k -> new Tuple2<>(sharingGroup, new
CoLocationGroup()));
+k -> new Tuple2<>(sharingGroup, new
CoLocationGroupImpl()));
Review comment:
See my answer above.
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:
us...@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #14584: [FLINK-20850][runtime] Removing usage of CoLocationConstraints
XComp commented on a change in pull request #14584:
URL: https://github.com/apache/flink/pull/14584#discussion_r556320074
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
##
@@ -425,12 +426,12 @@ public void setStrictlyCoLocatedWith(JobVertex
strictlyCoLocatedWith) {
"Strict co-location requires that both vertices are in the
same slot sharing group.");
}
-CoLocationGroup thisGroup = this.coLocationGroup;
-CoLocationGroup otherGroup = strictlyCoLocatedWith.coLocationGroup;
+CoLocationGroupImpl thisGroup = this.coLocationGroup;
+CoLocationGroupImpl otherGroup = strictlyCoLocatedWith.coLocationGroup;
Review comment:
The idea was to have a readonly interface as the colocation groups do
not need to be changed after initialization.
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:
us...@infra.apache.org
[jira] [Updated] (FLINK-20951) Flink SQL cli an exception will occur
[ https://issues.apache.org/jira/browse/FLINK-20951?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] YUJIANBO updated FLINK-20951: - Summary: Flink SQL cli an exception will occur (was: Flink SQL cli queries the partition table with hive as two fields. If the where condition does not specify all the two fields, an exception will occur) > Flink SQL cli an exception will occur > - > > Key: FLINK-20951 > URL: https://issues.apache.org/jira/browse/FLINK-20951 > Project: Flink > Issue Type: Bug > Components: Connectors / Hive, Table SQL / Runtime >Affects Versions: 1.12.0 > Environment: flink 1.12.0release-12 > sql-cli >Reporter: YUJIANBO >Priority: Major > > The production hive table is partitioned by two fields：datekey and event > I have do this test by Flink-sql-cli:(Spark Sql All is OK) > （1）First: > SELECT vid From table_A WHERE datekey = '20210112' AND event = 'XXX' AND vid > = 'aaaaaa'; (OK) > SELECT vid From table_A WHERE datekey = '20210112' AND vid = 'aa'; > (Error) > （2）Second: > SELECT vid From table_B WHERE datekey = '20210112' AND event = 'YYY' AND vid > = 'bb';(OK) > SELECT vid From table_B WHERE datekey = '20210112' AND vid = 'bb'; > (Error) > The exception is: > java.lang.RuntimeException: One or more fetchers have encountered exception > at > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:199) > at > org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:154) > at > org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:116) > at > org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:273) > at > org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:67) > at > org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) > at > org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:395) > at > org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:191) > at > org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:609) > at > org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:573) > at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755) > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570) > at java.lang.Thread.run(Thread.java:745) > Caused by: java.lang.RuntimeException: SplitFetcher thread 19 received > unexpected exception while polling the records > at > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:146) > at > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:101) > at > java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) > at java.util.concurrent.FutureTask.run(FutureTask.java:266) > at > java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) > at > java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) > ... 1 more > Caused by: java.lang.IllegalArgumentException > at java.nio.Buffer.position(Buffer.java:244) > at > org.apache.flink.hive.shaded.parquet.io.api.Binary$ByteBufferBackedBinary.getBytes(Binary.java:424) > at > org.apache.flink.hive.shaded.formats.parquet.vector.reader.BytesColumnReader.readBatchFromDictionaryIds(BytesColumnReader.java:79) > at > org.apache.flink.hive.shaded.formats.parquet.vector.reader.BytesColumnReader.readBatchFromDictionaryIds(BytesColumnReader.java:33) > at > org.apache.flink.hive.shaded.formats.parquet.vector.reader.AbstractColumnReader.readToVector(AbstractColumnReader.java:199) > at > org.apache.flink.hive.shaded.formats.parquet.ParquetVectorizedInputFormat$ParquetReader.nextBatch(ParquetVectorizedInputFormat.java:359) > at > org.apache.flink.hive.shaded.formats.parquet.ParquetVectorizedInputFormat$ParquetReader.readBatch(ParquetVectorizedInputFormat.java:328) > at > org.apache.flink.connector.file.src.impl.FileSourceSplitReader.fetch(FileSourceSplitReader.java:67) > at > org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:56) > at > org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:138) > ... 6 more -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot commented on pull request #14627: [FLINK-20946][python] Optimize Python ValueState Implementation In PyFlink
flinkbot commented on pull request #14627: URL: https://github.com/apache/flink/pull/14627#issuecomment-759268470 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 a0e6a39605a3d1a80f75f8764534dfb82d08c31f (Wed Jan 13 07:44:31 UTC 2021) **Warnings:** * No documentation files were touched! Remember to keep the Flink docs up to date! * **This pull request references an unassigned [Jira ticket](https://issues.apache.org/jira/browse/FLINK-20946).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work. Mention the bot in a comment to re-run the automated checks. ## 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. 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 Bot commands 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 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: us...@infra.apache.org
[jira] [Updated] (FLINK-20951) Flink SQL cli queries the partition table with hive as two fields. If the where condition does not specify all the two fields, an exception will occur
[ https://issues.apache.org/jira/browse/FLINK-20951?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] YUJIANBO updated FLINK-20951: - Description: The production hive table is partitioned by two fields：datekey and event I have do this test by Flink-sql-cli:(Spark Sql All is OK) （1）First: SELECT vid From table_A WHERE datekey = '20210112' AND event = 'XXX' AND vid = 'aa';(OK) SELECT vid From table_A WHERE datekey = '20210112' AND vid = 'aa'; (Error) （2）Second: SELECT vid From table_B WHERE datekey = '20210112' AND event = 'YYY' AND vid = 'bb';(OK) SELECT vid From table_B WHERE datekey = '20210112' AND vid = 'bb'; (Error) The exception is: java.lang.RuntimeException: One or more fetchers have encountered exception at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:199) at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:154) at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:116) at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:273) at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:67) at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:395) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:191) at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:609) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:573) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570) at java.lang.Thread.run(Thread.java:745) Caused by: java.lang.RuntimeException: SplitFetcher thread 19 received unexpected exception while polling the records at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:146) at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:101) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ... 1 more Caused by: java.lang.IllegalArgumentException at java.nio.Buffer.position(Buffer.java:244) at org.apache.flink.hive.shaded.parquet.io.api.Binary$ByteBufferBackedBinary.getBytes(Binary.java:424) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.BytesColumnReader.readBatchFromDictionaryIds(BytesColumnReader.java:79) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.BytesColumnReader.readBatchFromDictionaryIds(BytesColumnReader.java:33) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.AbstractColumnReader.readToVector(AbstractColumnReader.java:199) at org.apache.flink.hive.shaded.formats.parquet.ParquetVectorizedInputFormat$ParquetReader.nextBatch(ParquetVectorizedInputFormat.java:359) at org.apache.flink.hive.shaded.formats.parquet.ParquetVectorizedInputFormat$ParquetReader.readBatch(ParquetVectorizedInputFormat.java:328) at org.apache.flink.connector.file.src.impl.FileSourceSplitReader.fetch(FileSourceSplitReader.java:67) at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:56) at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:138) ... 6 more was: The production hive table is partitioned by two fields：datekey and event I have do this test by Flink-sql-cli:(Spark Sql All is OK) （1）First: SELECT vid From table_A WHERE datekey = '20210112' AND event = 'XXX' AND vid = 'aa';(OK) SELECT vid From table_A WHERE datekey = '20210112' AND vid = 'aa'; (Error) （2）Second: SELECT vid From table_B WHERE datekey = '20210112' AND event = 'YYY' AND vid = 'bb';(OK) SELECT vid From table_B WHERE datekey = '20210112' AND vid = 'bb'; (Error) The exception is: java.lang.RuntimeException: One or more fetchers have encountered exception at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:199) at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:154) at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext
[jira] [Updated] (FLINK-20946) Optimize Python ValueState Implementation In PyFlink
[ https://issues.apache.org/jira/browse/FLINK-20946?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-20946: --- Labels: pull-request-available (was: ) > Optimize Python ValueState Implementation In PyFlink > > > Key: FLINK-20946 > URL: https://issues.apache.org/jira/browse/FLINK-20946 > Project: Flink > Issue Type: Improvement > Components: API / Python >Affects Versions: 1.13.0 >Reporter: Huang Xingbo >Priority: Major > Labels: pull-request-available > Fix For: 1.13.0 > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] xiaoHoly commented on a change in pull request #14616: [FLINK-20879][table-api] Use MemorySize type instead of String type for memory ConfigOption in ExecutionConfigOptions
xiaoHoly commented on a change in pull request #14616:
URL: https://github.com/apache/flink/pull/14616#discussion_r556312510
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/utils/ExecNodeUtil.java
##
@@ -42,8 +42,8 @@
*
* TODO: This method can be removed once FLINK-20879 is finished.
*/
-public static long getMemorySize(TableConfig tableConfig,
ConfigOption option) {
-return
MemorySize.parse(tableConfig.getConfiguration().getString(option)).getBytes();
+public static long getMemorySize(TableConfig tableConfig,
ConfigOption option) {
Review comment:
okey~
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:
us...@infra.apache.org
[GitHub] [flink] HuangXingBo opened a new pull request #14627: [FLINK-20946][python] Optimize Python ValueState Implementation In PyFlink
HuangXingBo opened a new pull request #14627: URL: https://github.com/apache/flink/pull/14627 ## What is the purpose of the change *This pull request will Optimize Python ValueState Implementation In PyFlink* ## Brief change log - *Optimize Python ValueState Implementation In PyFlink* ## Verifying this change This change added tests and can be verified as follows: - *Original Tests* ## 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? (no) - If yes, how is the feature documented? (not applicable) 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: us...@infra.apache.org
[jira] [Commented] (FLINK-20900) Extend documentation guidelines to cover formatting of commands
[ https://issues.apache.org/jira/browse/FLINK-20900?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263957#comment-17263957 ] Robert Metzger commented on FLINK-20900: Merged to flink-web in https://github.com/apache/flink-web/commit/38b3a48ca4ea40b2768f49e1c1df181c276b076e > Extend documentation guidelines to cover formatting of commands > --- > > Key: FLINK-20900 > URL: https://issues.apache.org/jira/browse/FLINK-20900 > Project: Flink > Issue Type: Improvement > Components: Documentation >Reporter: Matthias >Assignee: Matthias >Priority: Major > Labels: pull-request-available > > As part of refactoring the deployment documentation we came up with some > agreements that we could add to the docs guidelines to get a more consistent > look & feel: > * long parameter names for commands > * one parameter per line > * parameter line indented > * command preceded by `$` -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Closed] (FLINK-20900) Extend documentation guidelines to cover formatting of commands
[ https://issues.apache.org/jira/browse/FLINK-20900?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Robert Metzger closed FLINK-20900. -- Resolution: Fixed > Extend documentation guidelines to cover formatting of commands > --- > > Key: FLINK-20900 > URL: https://issues.apache.org/jira/browse/FLINK-20900 > Project: Flink > Issue Type: Improvement > Components: Documentation >Reporter: Matthias >Assignee: Matthias >Priority: Major > Labels: pull-request-available > > As part of refactoring the deployment documentation we came up with some > agreements that we could add to the docs guidelines to get a more consistent > look & feel: > * long parameter names for commands > * one parameter per line > * parameter line indented > * command preceded by `$` -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] rmetzger commented on a change in pull request #14584: [FLINK-20850][runtime] Removing usage of CoLocationConstraints
rmetzger commented on a change in pull request #14584:
URL: https://github.com/apache/flink/pull/14584#discussion_r556306735
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobVertex.java
##
@@ -425,12 +426,12 @@ public void setStrictlyCoLocatedWith(JobVertex
strictlyCoLocatedWith) {
"Strict co-location requires that both vertices are in the
same slot sharing group.");
}
-CoLocationGroup thisGroup = this.coLocationGroup;
-CoLocationGroup otherGroup = strictlyCoLocatedWith.coLocationGroup;
+CoLocationGroupImpl thisGroup = this.coLocationGroup;
+CoLocationGroupImpl otherGroup = strictlyCoLocatedWith.coLocationGroup;
Review comment:
Wouldn't it be cleaner to add the `addVertex()` method to the
`CoLocationGroup` interface?
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
##
@@ -1018,10 +1018,10 @@ private void setCoLocation() {
"Cannot use a co-location constraint without a
slot sharing group");
}
-Tuple2 constraint =
+Tuple2 constraint =
coLocationGroups.computeIfAbsent(
coLocationGroupKey,
-k -> new Tuple2<>(sharingGroup, new
CoLocationGroup()));
+k -> new Tuple2<>(sharingGroup, new
CoLocationGroupImpl()));
Review comment:
Adding the addVertex method to the interface would not require touching
this test
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:
us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14604: [FLINK-20861][format/json] Provide an option for serializing DECIMALs in JSON as plain number instead of scientific notation
flinkbot edited a comment on pull request #14604: URL: https://github.com/apache/flink/pull/14604#issuecomment-757521857 ## CI report: * 6f16ecba1741139347662f1a892cca2885b603c9 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11862) * 34a633e801bcb8100e0962b358b4e400a47c7047 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11959) * 095aa26dba7a04800cb203e252774d0d7655cc25 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11962) * 14ecbcb1bc4fc86bcdda844a48b8c187f6c9a96a UNKNOWN Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14612: [FLINK-20864][runtime] Introduce the DEFAULT resource profile for the…
flinkbot edited a comment on pull request #14612: URL: https://github.com/apache/flink/pull/14612#issuecomment-758375136 ## CI report: * e8f754cc415e4ff814616ed4eb1c7160cb765d7d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11952) * 1617d36fad644d8fc50bb0a8d9e6a976eca942f0 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11965) Bot commands 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 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: us...@infra.apache.org
[jira] [Comment Edited] (FLINK-20919) After the flink batch job is completed, the yarn application cannot be completed.
[
https://issues.apache.org/jira/browse/FLINK-20919?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263954#comment-17263954
]
Wu edited comment on FLINK-20919 at 1/13/21, 7:30 AM:
--
[~fly_in_gis] Try to run the code as follows. Use the filesystem and print
connector
{code:java}
//代码占位符
EnvironmentSettings settings =
EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
TableEnvironment tableEnv = TableEnvironment.create(settings);
tableEnv.executeSql("create table file_table");
tableEnv.executeSql("create table print_table"); String sql = "select count(1)
from file_table";
Table table = tableEnv.sqlQuery(sql);
tableEnv.createTemporaryView("t", table);
tableEnv.from("t").executeInsert("print_table");
{code}
was (Author: janze):
[~fly_in_gis] Try to run the code as follows. Use the filesystem and print
connector
EnvironmentSettings settings =
EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
TableEnvironment tableEnv = TableEnvironment.create(settings);
tableEnv.executeSql("create table file_table");
tableEnv.executeSql("create table print_table");String sql =
"select count(1) from file_table";
Table table = tableEnv.sqlQuery(sql);
tableEnv.createTemporaryView("t", table);
tableEnv.from("t").executeInsert("print_table");
> After the flink batch job is completed, the yarn application cannot be
> completed.
> -
>
> Key: FLINK-20919
> URL: https://issues.apache.org/jira/browse/FLINK-20919
> Project: Flink
> Issue Type: Bug
> Components: Deployment / YARN
>Affects Versions: 1.11.2
>Reporter: Wu
>Priority: Major
> Attachments: flink-Overview.png, flink-completed.png,
> flink-jobManager.png, kaLr8Coy.png
>
>
> I submit flink batch job in yarn-cluster mode. After the flink batch job is
> completed, the yarn application cannot be completed. The yarn application
> still occupies a vcore. How to automatically close the yarn application.
>
> {code:java}
> //代码占位符
> EnvironmentSettings settings =
> EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
> TableEnvironment tableEnv = TableEnvironment.create(settings);
> tableEnv.executeSql("create table file_table");
> tableEnv.executeSql("create table print_table");
> String sql = "select count(1) from file_table";
> Table table = tableEnv.sqlQuery(sql);
> tableEnv.createTemporaryView("t", table);
> tableEnv.from("t").executeInsert("print_table");
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-20919) After the flink batch job is completed, the yarn application cannot be completed.
[
https://issues.apache.org/jira/browse/FLINK-20919?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263954#comment-17263954
]
Wu commented on FLINK-20919:
[~fly_in_gis] Try to run the code as follows. Use the filesystem and print
connector
EnvironmentSettings settings =
EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
TableEnvironment tableEnv = TableEnvironment.create(settings);
tableEnv.executeSql("create table file_table");
tableEnv.executeSql("create table print_table");String sql =
"select count(1) from file_table";
Table table = tableEnv.sqlQuery(sql);
tableEnv.createTemporaryView("t", table);
tableEnv.from("t").executeInsert("print_table");
> After the flink batch job is completed, the yarn application cannot be
> completed.
> -
>
> Key: FLINK-20919
> URL: https://issues.apache.org/jira/browse/FLINK-20919
> Project: Flink
> Issue Type: Bug
> Components: Deployment / YARN
>Affects Versions: 1.11.2
>Reporter: Wu
>Priority: Major
> Attachments: flink-Overview.png, flink-completed.png,
> flink-jobManager.png, kaLr8Coy.png
>
>
> I submit flink batch job in yarn-cluster mode. After the flink batch job is
> completed, the yarn application cannot be completed. The yarn application
> still occupies a vcore. How to automatically close the yarn application.
>
> {code:java}
> //代码占位符
> EnvironmentSettings settings =
> EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
> TableEnvironment tableEnv = TableEnvironment.create(settings);
> tableEnv.executeSql("create table file_table");
> tableEnv.executeSql("create table print_table");
> String sql = "select count(1) from file_table";
> Table table = tableEnv.sqlQuery(sql);
> tableEnv.createTemporaryView("t", table);
> tableEnv.from("t").executeInsert("print_table");
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot commented on pull request #14626: [FLINK-20948][table-planner-blink] Introduce StreamPhysicalDeduplicate, and make StreamExecDeduplicate only extended from ExecNode
flinkbot commented on pull request #14626: URL: https://github.com/apache/flink/pull/14626#issuecomment-759262176 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 1387747f865a79d120ee7f5d6f24685c90e0076f (Wed Jan 13 07:29:59 UTC 2021) **Warnings:** * No documentation files were touched! Remember to keep the Flink docs up to date! Mention the bot in a comment to re-run the automated checks. ## 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. 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 Bot commands 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 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: us...@infra.apache.org
[jira] [Updated] (FLINK-20951) Flink SQL cli queries the partition table with hive as two fields. If the where condition does not specify all the two fields, an exception will occur
[ https://issues.apache.org/jira/browse/FLINK-20951?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] YUJIANBO updated FLINK-20951: - Description: The production hive table is partitioned by two fields：datekey and event I have do this test by Flink-sql-cli:(Spark Sql All is OK) （1）First: SELECT vid From table_A WHERE datekey = '20210112' AND event = 'XXX' AND vid = 'aa';(OK) SELECT vid From table_A WHERE datekey = '20210112' AND vid = 'aa'; (Error) （2）Second: SELECT vid From table_B WHERE datekey = '20210112' AND event = 'YYY' AND vid = 'bb';(OK) SELECT vid From table_B WHERE datekey = '20210112' AND vid = 'bb'; (Error) The exception is: java.lang.RuntimeException: One or more fetchers have encountered exception at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:199) at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:154) at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:116) at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:273) at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:67) at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:395) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:191) at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:609) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:573) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570) at java.lang.Thread.run(Thread.java:745) Caused by: java.lang.RuntimeException: SplitFetcher thread 19 received unexpected exception while polling the records at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:146) at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:101) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ... 1 more Caused by: java.lang.IllegalArgumentException at java.nio.Buffer.position(Buffer.java:244) at org.apache.flink.hive.shaded.parquet.io.api.Binary$ByteBufferBackedBinary.getBytes(Binary.java:424) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.BytesColumnReader.readBatchFromDictionaryIds(BytesColumnReader.java:79) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.BytesColumnReader.readBatchFromDictionaryIds(BytesColumnReader.java:33) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.AbstractColumnReader.readToVector(AbstractColumnReader.java:199) at org.apache.flink.hive.shaded.formats.parquet.ParquetVectorizedInputFormat$ParquetReader.nextBatch(ParquetVectorizedInputFormat.java:359) at org.apache.flink.hive.shaded.formats.parquet.ParquetVectorizedInputFormat$ParquetReader.readBatch(ParquetVectorizedInputFormat.java:328) at org.apache.flink.connector.file.src.impl.FileSourceSplitReader.fetch(FileSourceSplitReader.java:67) at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:56) at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:138) ... 6 more Flink Email: http://apache-flink.147419.n8.nabble.com/Flink-sql-cli-hive-where-td10139.html was: The production hive table is partitioned by two fields：datekey and event I have do this test:(Spark Sql All is OK) （1）First: SELECT vid From table_A WHERE datekey = '20210112' AND event = 'XXX' AND vid = 'aa';(OK) SELECT vid From table_A WHERE datekey = '20210112' AND vid = 'aa'; (Error) （2）Second: SELECT vid From table_B WHERE datekey = '20210112' AND event = 'YYY' AND vid = 'bb';(OK) SELECT vid From table_B WHERE datekey = '20210112' AND vid = 'bb'; (Error) The exception is: java.lang.RuntimeException: One or more fetchers have encountered exception at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:199) at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:154
[GitHub] [flink] wuchong merged pull request #14617: [hotfix][doc] Fix default format in document for temporal function FROM_UNIXTIME(numeric, format)
wuchong merged pull request #14617: URL: https://github.com/apache/flink/pull/14617 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: us...@infra.apache.org
[jira] [Updated] (FLINK-20951) Flink SQL cli queries the partition table with hive as two fields. If the where condition does not specify all the two fields, an exception will occur
[ https://issues.apache.org/jira/browse/FLINK-20951?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] YUJIANBO updated FLINK-20951: - Description: The production hive table is partitioned by two fields：datekey and event I have do this test:(Spark Sql All is OK) （1）First: SELECT vid From table_A WHERE datekey = '20210112' AND event = 'XXX' AND vid = 'aa';(OK) SELECT vid From table_A WHERE datekey = '20210112' AND vid = 'aa'; (Error) （2）Second: SELECT vid From table_B WHERE datekey = '20210112' AND event = 'YYY' AND vid = 'bb';(OK) SELECT vid From table_B WHERE datekey = '20210112' AND vid = 'bb'; (Error) The exception is: java.lang.RuntimeException: One or more fetchers have encountered exception at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:199) at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:154) at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:116) at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:273) at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:67) at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:395) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:191) at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:609) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:573) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570) at java.lang.Thread.run(Thread.java:745) Caused by: java.lang.RuntimeException: SplitFetcher thread 19 received unexpected exception while polling the records at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:146) at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:101) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ... 1 more Caused by: java.lang.IllegalArgumentException at java.nio.Buffer.position(Buffer.java:244) at org.apache.flink.hive.shaded.parquet.io.api.Binary$ByteBufferBackedBinary.getBytes(Binary.java:424) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.BytesColumnReader.readBatchFromDictionaryIds(BytesColumnReader.java:79) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.BytesColumnReader.readBatchFromDictionaryIds(BytesColumnReader.java:33) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.AbstractColumnReader.readToVector(AbstractColumnReader.java:199) at org.apache.flink.hive.shaded.formats.parquet.ParquetVectorizedInputFormat$ParquetReader.nextBatch(ParquetVectorizedInputFormat.java:359) at org.apache.flink.hive.shaded.formats.parquet.ParquetVectorizedInputFormat$ParquetReader.readBatch(ParquetVectorizedInputFormat.java:328) at org.apache.flink.connector.file.src.impl.FileSourceSplitReader.fetch(FileSourceSplitReader.java:67) at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:56) at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:138) ... 6 more Flink Email: http://apache-flink.147419.n8.nabble.com/Flink-sql-cli-hive-where-td10139.html was: The production hive table is partitioned by two fields：datekey and event I have do this test: （1）First: SELECT vid From table_A WHERE datekey = '20210112' AND event = 'XXX' AND vid = 'aa';(OK) SELECT vid From table_A WHERE datekey = '20210112' AND vid = 'aa'; (Error) （2）Second: SELECT vid From table_B WHERE datekey = '20210112' AND event = 'YYY' AND vid = 'bb';(OK) SELECT vid From table_B WHERE datekey = '20210112' AND vid = 'bb'; (Error) The exception is: java.lang.RuntimeException: One or more fetchers have encountered exception at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:199) at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:154
[GitHub] [flink] godfreyhe opened a new pull request #14626: [FLINK-20948][table-planner-blink] Introduce StreamPhysicalDeduplicate, and make StreamExecDeduplicate only extended from ExecNode
godfreyhe opened a new pull request #14626: URL: https://github.com/apache/flink/pull/14626 ## What is the purpose of the change *Separate the implementation of StreamExecDeduplicate* ## Brief change log - *Introduce StreamPhysicalDeduplicate, and make StreamExecDeduplicate only extended from ExecNode* ## Verifying this change This change is a refactoring rework covered by existing tests. ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (yes / **no**) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**) - The serializers: (yes / **no** / don't know) - The runtime per-record code paths (performance sensitive): (yes / **no** / don't know) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (yes / **no** / don't know) - The S3 file system connector: (yes / **no** / don't know) ## Documentation - Does this pull request introduce a new feature? (yes / **no**) - If yes, how is the feature documented? (not applicable / docs / JavaDocs / **not documented**) 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: us...@infra.apache.org
[jira] [Updated] (FLINK-20948) Separate the implementation of StreamExecDeduplicate
[ https://issues.apache.org/jira/browse/FLINK-20948?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-20948: --- Labels: pull-request-available (was: ) > Separate the implementation of StreamExecDeduplicate > > > Key: FLINK-20948 > URL: https://issues.apache.org/jira/browse/FLINK-20948 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Reporter: godfrey he >Assignee: godfrey he >Priority: Major > Labels: pull-request-available > Fix For: 1.13.0 > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] lmagic233 commented on pull request #14604: [FLINK-20861][format/json] Provide an option for serializing DECIMALs in JSON as plain number instead of scientific notation
lmagic233 commented on pull request #14604: URL: https://github.com/apache/flink/pull/14604#issuecomment-759259586 @wuchong Squashed commits and optimized test code naming & documentations according to your revision, please verify. 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: us...@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #14616: [FLINK-20879][table-api] Use MemorySize type instead of String type for memory ConfigOption in ExecutionConfigOptions
wuchong commented on a change in pull request #14616:
URL: https://github.com/apache/flink/pull/14616#discussion_r556307216
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/batch/BatchPhysicalSortMergeJoin.scala
##
@@ -189,7 +189,7 @@ class BatchPhysicalSortMergeJoin(
rightAllKey,
filterNulls,
condition,
- estimateOutputSize(getLeft) < estimateOutputSize(getRight),
+ estimateOutputSize(getLeft) < estimateOutputSize(getRight),
Review comment:
Please revert this.
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/utils/ExecNodeUtil.java
##
@@ -42,8 +42,8 @@
*
* TODO: This method can be removed once FLINK-20879 is finished.
*/
-public static long getMemorySize(TableConfig tableConfig,
ConfigOption option) {
-return
MemorySize.parse(tableConfig.getConfiguration().getString(option)).getBytes();
+public static long getMemorySize(TableConfig tableConfig,
ConfigOption option) {
Review comment:
I think we don't need this utility method anymore.
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:
us...@infra.apache.org
[jira] [Commented] (FLINK-20943) AsyncCheckpointRunnable.cleanup() runs in the main task thread
[
https://issues.apache.org/jira/browse/FLINK-20943?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263934#comment-17263934
]
Yun Tang commented on FLINK-20943:
--
There actually has several operations in
{{SubtaskCheckpointCoordinatorImpl#notifyCheckpointAborted}}, what's your plan
to make several of them as async or just {{AsyncCheckpointRunnable.cleanup()}}
as async?
> AsyncCheckpointRunnable.cleanup() runs in the main task thread
> --
>
> Key: FLINK-20943
> URL: https://issues.apache.org/jira/browse/FLINK-20943
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Checkpointing
>Affects Versions: 1.12.0, 1.11.3, 1.13.0
>Reporter: Roman Khachatryan
>Assignee: Roman Khachatryan
>Priority: Major
> Fix For: 1.13.0, 1.12.2
>
>
> In FLINK-8871 an RPC notification about checkpoint abortion was added.
> Upon receiving it, task cancels the running state-persisting futures. It does
> so through the mailbox.
> Cancelling the futures may involve waiting for their result (if they can't be
> cancelled).
>
> This essentially makes checkpoint sync from async upon receiving such RPC.
>
> cc [~yunta], [~pnowojski]
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] xiaoHoly commented on a change in pull request #14531: [FLINK-20777][Connector][Kafka] Property "partition.discovery.interval.ms" shoule be enabled by default for unbounded mode, and
xiaoHoly commented on a change in pull request #14531:
URL: https://github.com/apache/flink/pull/14531#discussion_r556309253
##
File path:
flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java
##
@@ -458,6 +458,29 @@ private boolean maybeOverride(String key, String value,
boolean override) {
return overridden;
}
+private boolean maybeOverridePartitionDiscovery(String key, String value,
boolean override) {
Review comment:
Hi,@becketqin.Does this unit test only need to check the properties of
KafkaSource or partitionDiscoveryIntervalMs of KafkaSourceEnumerator?
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:
us...@infra.apache.org
[jira] [Issue Comment Deleted] (FLINK-20885) Exception when use 'canal-json.table.include' to filter Canal binlog but table contains 'source' column
[ https://issues.apache.org/jira/browse/FLINK-20885?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Nicholas Jiang updated FLINK-20885: --- Comment: was deleted (was: [~wangfeiair2324], I operates the steps above you mentioned, but don't reproduce the problem you mentioned. Could you please add Wechat 834031959 to offline discussion.) > Exception when use 'canal-json.table.include' to filter Canal binlog but > table contains 'source' column > --- > > Key: FLINK-20885 > URL: https://issues.apache.org/jira/browse/FLINK-20885 > Project: Flink > Issue Type: Bug > Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table > SQL / Ecosystem >Reporter: Jark Wu >Assignee: Nicholas Jiang >Priority: Major > Attachments: image-2021-01-08-14-52-53-443.png, > image-2021-01-08-14-53-40-476.png, image-2021-01-13-15-02-33-266.png, > screenshot-1.png > > > I found a bug in the canal code. 'canal-json.table.include' does not filter > out the binlog of the specified table correctly, which will cause an error in > the parsing section. For example, if I want to read the binlog of > canal-json.table.include = 'a' table, there is a source field of int in table > a, but at this time if table b also has a source field of string, An error > will be reported. > !screenshot-1.png! -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #14612: [FLINK-20864][runtime] Introduce the DEFAULT resource profile for the…
flinkbot edited a comment on pull request #14612: URL: https://github.com/apache/flink/pull/14612#issuecomment-758375136 ## CI report: * e8f754cc415e4ff814616ed4eb1c7160cb765d7d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11952) * 1617d36fad644d8fc50bb0a8d9e6a976eca942f0 UNKNOWN Bot commands 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 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: us...@infra.apache.org
[jira] [Commented] (FLINK-20885) Exception when use 'canal-json.table.include' to filter Canal binlog but table contains 'source' column
[ https://issues.apache.org/jira/browse/FLINK-20885?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263929#comment-17263929 ] Nicholas Jiang commented on FLINK-20885: [~wangfeiair2324], I operates the steps above you mentioned, but don't reproduce the problem you mentioned. Could you please add Wechat 834031959 to offline discussion. > Exception when use 'canal-json.table.include' to filter Canal binlog but > table contains 'source' column > --- > > Key: FLINK-20885 > URL: https://issues.apache.org/jira/browse/FLINK-20885 > Project: Flink > Issue Type: Bug > Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table > SQL / Ecosystem >Reporter: Jark Wu >Assignee: Nicholas Jiang >Priority: Major > Attachments: image-2021-01-08-14-52-53-443.png, > image-2021-01-08-14-53-40-476.png, image-2021-01-13-15-02-33-266.png, > screenshot-1.png > > > I found a bug in the canal code. 'canal-json.table.include' does not filter > out the binlog of the specified table correctly, which will cause an error in > the parsing section. For example, if I want to read the binlog of > canal-json.table.include = 'a' table, there is a source field of int in table > a, but at this time if table b also has a source field of string, An error > will be reported. > !screenshot-1.png! -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #14606: [Flink-20876][table-planner-blink] Separate the implementation of StreamExecTemporalJoin
flinkbot edited a comment on pull request #14606: URL: https://github.com/apache/flink/pull/14606#issuecomment-757754275 ## CI report: * 3776b52cfe3535dcc193b3a922a7d1d658126d66 UNKNOWN * 155b18c169e45a97cd52c5b43883d5cf6b79f038 UNKNOWN * b6ca209812ae0e584d738f0ea6775a525a624a2f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11956) * 9276dee92aa80400b79a80168d19553aa2689a10 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11964) Bot commands 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 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: us...@infra.apache.org
[jira] [Commented] (FLINK-20454) Allow to read metadata for debezium-avro-confluent format
[
https://issues.apache.org/jira/browse/FLINK-20454?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263928#comment-17263928
]
Jark Wu commented on FLINK-20454:
-
Can't {{source.properties}} be read as a map type? [~hailong wang]
> Allow to read metadata for debezium-avro-confluent format
> -
>
> Key: FLINK-20454
> URL: https://issues.apache.org/jira/browse/FLINK-20454
> Project: Flink
> Issue Type: Sub-task
> Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table
> SQL / Ecosystem
>Reporter: Jark Wu
>Assignee: hailong wang
>Priority: Major
> Fix For: 1.13.0
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Comment Edited] (FLINK-20885) Exception when use 'canal-json.table.include' to filter Canal binlog but table contains 'source' column
[
https://issues.apache.org/jira/browse/FLINK-20885?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263923#comment-17263923
]
wangfei edited comment on FLINK-20885 at 1/13/21, 7:03 AM:
---
Remove \, the JIRA automatically adds \
!image-2021-01-13-15-02-33-266.png!
was (Author: wangfeiair2324):
{"data":[\{"id":"0","username":"zhangsna","password":"passwd","AddTime":"2020-12-25
14:23:07"}],"database":"plink","es":1610520384000,"id":92151,"isDdl":false,"mysqlType":\{"id":"int(11)","username":"varchar(255)","password":"varchar(255)","AddTime":"datetime"},"old":[\{"password":""}],"pkNames":["id"],"sql":"","sqlType":\{"id":4,"username":12,"password":12,"AddTime":93},"table":"test_1","ts":1610520384579,"type":"UPDATE"}
> Exception when use 'canal-json.table.include' to filter Canal binlog but
> table contains 'source' column
> ---
>
> Key: FLINK-20885
> URL: https://issues.apache.org/jira/browse/FLINK-20885
> Project: Flink
> Issue Type: Bug
> Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table
> SQL / Ecosystem
>Reporter: Jark Wu
>Assignee: Nicholas Jiang
>Priority: Major
> Attachments: image-2021-01-08-14-52-53-443.png,
> image-2021-01-08-14-53-40-476.png, image-2021-01-13-15-02-33-266.png,
> screenshot-1.png
>
>
> I found a bug in the canal code. 'canal-json.table.include' does not filter
> out the binlog of the specified table correctly, which will cause an error in
> the parsing section. For example, if I want to read the binlog of
> canal-json.table.include = 'a' table, there is a source field of int in table
> a, but at this time if table b also has a source field of string, An error
> will be reported.
> !screenshot-1.png!
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-20885) Exception when use 'canal-json.table.include' to filter Canal binlog but table contains 'source' column
[
https://issues.apache.org/jira/browse/FLINK-20885?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263923#comment-17263923
]
wangfei commented on FLINK-20885:
-
{"data":[\{"id":"0","username":"zhangsna","password":"passwd","AddTime":"2020-12-25
14:23:07"}],"database":"plink","es":1610520384000,"id":92151,"isDdl":false,"mysqlType":\{"id":"int(11)","username":"varchar(255)","password":"varchar(255)","AddTime":"datetime"},"old":[\{"password":""}],"pkNames":["id"],"sql":"","sqlType":\{"id":4,"username":12,"password":12,"AddTime":93},"table":"test_1","ts":1610520384579,"type":"UPDATE"}
> Exception when use 'canal-json.table.include' to filter Canal binlog but
> table contains 'source' column
> ---
>
> Key: FLINK-20885
> URL: https://issues.apache.org/jira/browse/FLINK-20885
> Project: Flink
> Issue Type: Bug
> Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table
> SQL / Ecosystem
>Reporter: Jark Wu
>Assignee: Nicholas Jiang
>Priority: Major
> Attachments: image-2021-01-08-14-52-53-443.png,
> image-2021-01-08-14-53-40-476.png, screenshot-1.png
>
>
> I found a bug in the canal code. 'canal-json.table.include' does not filter
> out the binlog of the specified table correctly, which will cause an error in
> the parsing section. For example, if I want to read the binlog of
> canal-json.table.include = 'a' table, there is a source field of int in table
> a, but at this time if table b also has a source field of string, An error
> will be reported.
> !screenshot-1.png!
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (FLINK-20952) Changelog json formats should support inherit options from JSON format
[ https://issues.apache.org/jira/browse/FLINK-20952?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jark Wu updated FLINK-20952: Issue Type: Improvement (was: Bug) > Changelog json formats should support inherit options from JSON format > -- > > Key: FLINK-20952 > URL: https://issues.apache.org/jira/browse/FLINK-20952 > Project: Flink > Issue Type: Improvement > Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table > SQL / Ecosystem >Reporter: Jark Wu >Priority: Major > Fix For: 1.13.0 > > > Recently, we introduced several config options for json format, e.g. > FLINK-20861. It reveals a potential problem that adding a small config option > into json may need touch debezium-json, canal-json, maxwell-json formats. > This is verbose and error-prone. We need an abstract machanism support > reuable options. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-20952) Changelog json formats should support inherit options from JSON format
Jark Wu created FLINK-20952: --- Summary: Changelog json formats should support inherit options from JSON format Key: FLINK-20952 URL: https://issues.apache.org/jira/browse/FLINK-20952 Project: Flink Issue Type: Bug Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table SQL / Ecosystem Reporter: Jark Wu Fix For: 1.13.0 Recently, we introduced several config options for json format, e.g. FLINK-20861. It reveals a potential problem that adding a small config option into json may need touch debezium-json, canal-json, maxwell-json formats. This is verbose and error-prone. We need an abstract machanism support reuable options. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Comment Edited] (FLINK-20885) Exception when use 'canal-json.table.include' to filter Canal binlog but table contains 'source' column
[ https://issues.apache.org/jira/browse/FLINK-20885?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263921#comment-17263921 ] Nicholas Jiang edited comment on FLINK-20885 at 1/13/21, 6:59 AM: -- [~wangfeiair2324], I have already used https://www.bejson.com to check your JSON above provided. But unfortunately, this format of JSON above checks fails. was (Author: nicholasjiang): [~wangfeiair2324], do you confirm that the JSON above could be parsed? You could use https://www.bejson.com to check your JSON. > Exception when use 'canal-json.table.include' to filter Canal binlog but > table contains 'source' column > --- > > Key: FLINK-20885 > URL: https://issues.apache.org/jira/browse/FLINK-20885 > Project: Flink > Issue Type: Bug > Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table > SQL / Ecosystem >Reporter: Jark Wu >Assignee: Nicholas Jiang >Priority: Major > Attachments: image-2021-01-08-14-52-53-443.png, > image-2021-01-08-14-53-40-476.png, screenshot-1.png > > > I found a bug in the canal code. 'canal-json.table.include' does not filter > out the binlog of the specified table correctly, which will cause an error in > the parsing section. For example, if I want to read the binlog of > canal-json.table.include = 'a' table, there is a source field of int in table > a, but at this time if table b also has a source field of string, An error > will be reported. > !screenshot-1.png! -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] KarmaGYZ commented on pull request #14612: [FLINK-20864][runtime] Introduce the DEFAULT resource profile for the…
KarmaGYZ commented on pull request #14612: URL: https://github.com/apache/flink/pull/14612#issuecomment-759248904 Thanks for the review, @xintongsong . PR updated. 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: us...@infra.apache.org
[jira] [Updated] (FLINK-20951) Flink SQL cli queries the partition table with hive as two fields. If the where condition does not specify all the two fields, an exception will occur
[ https://issues.apache.org/jira/browse/FLINK-20951?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] YUJIANBO updated FLINK-20951: - Description: The production hive table is partitioned by two fields：datekey and event I have do this test: （1）First: SELECT vid From table_A WHERE datekey = '20210112' AND event = 'XXX' AND vid = 'aa';(OK) SELECT vid From table_A WHERE datekey = '20210112' AND vid = 'aa'; (Error) （2）Second: SELECT vid From table_B WHERE datekey = '20210112' AND event = 'YYY' AND vid = 'bb';(OK) SELECT vid From table_B WHERE datekey = '20210112' AND vid = 'bb'; (Error) The exception is: java.lang.RuntimeException: One or more fetchers have encountered exception at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:199) at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:154) at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:116) at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:273) at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:67) at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:395) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:191) at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:609) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:573) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570) at java.lang.Thread.run(Thread.java:745) Caused by: java.lang.RuntimeException: SplitFetcher thread 19 received unexpected exception while polling the records at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:146) at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:101) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ... 1 more Caused by: java.lang.IllegalArgumentException at java.nio.Buffer.position(Buffer.java:244) at org.apache.flink.hive.shaded.parquet.io.api.Binary$ByteBufferBackedBinary.getBytes(Binary.java:424) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.BytesColumnReader.readBatchFromDictionaryIds(BytesColumnReader.java:79) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.BytesColumnReader.readBatchFromDictionaryIds(BytesColumnReader.java:33) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.AbstractColumnReader.readToVector(AbstractColumnReader.java:199) at org.apache.flink.hive.shaded.formats.parquet.ParquetVectorizedInputFormat$ParquetReader.nextBatch(ParquetVectorizedInputFormat.java:359) at org.apache.flink.hive.shaded.formats.parquet.ParquetVectorizedInputFormat$ParquetReader.readBatch(ParquetVectorizedInputFormat.java:328) at org.apache.flink.connector.file.src.impl.FileSourceSplitReader.fetch(FileSourceSplitReader.java:67) at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:56) at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:138) ... 6 more Flink Email: http://apache-flink.147419.n8.nabble.com/Flink-sql-cli-hive-where-td10139.html was: The production hive table is partitioned by two fields：datekey and event I have do this test: （1）First: SELECT vid From table_A WHERE datekey = '20210112' AND event = 'XXX' AND vid = 'aa';(OK) SELECT vid From table_A WHERE datekey = '20210112' AND vid = 'aa'; (Error) （2）第二组对比 SELECT vid From table_B WHERE datekey = '20210112' AND event = 'YYY' AND vid = 'bb';(OK) SELECT vid From table_B WHERE datekey = '20210112' AND vid = 'bb'; (Error) The exception is: java.lang.RuntimeException: One or more fetchers have encountered exception at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:199) at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:154
[jira] [Commented] (FLINK-20885) Exception when use 'canal-json.table.include' to filter Canal binlog but table contains 'source' column
[ https://issues.apache.org/jira/browse/FLINK-20885?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263921#comment-17263921 ] Nicholas Jiang commented on FLINK-20885: [~wangfeiair2324], do you confirm that the JSON above could be parsed? You could use https://www.bejson.com to check your JSON. > Exception when use 'canal-json.table.include' to filter Canal binlog but > table contains 'source' column > --- > > Key: FLINK-20885 > URL: https://issues.apache.org/jira/browse/FLINK-20885 > Project: Flink > Issue Type: Bug > Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table > SQL / Ecosystem >Reporter: Jark Wu >Assignee: Nicholas Jiang >Priority: Major > Attachments: image-2021-01-08-14-52-53-443.png, > image-2021-01-08-14-53-40-476.png, screenshot-1.png > > > I found a bug in the canal code. 'canal-json.table.include' does not filter > out the binlog of the specified table correctly, which will cause an error in > the parsing section. For example, if I want to read the binlog of > canal-json.table.include = 'a' table, there is a source field of int in table > a, but at this time if table b also has a source field of string, An error > will be reported. > !screenshot-1.png! -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-20951) Flink SQL cli queries the partition table with hive as two fields. If the where condition does not specify all the two fields, an exception will occur
YUJIANBO created FLINK-20951: Summary: Flink SQL cli queries the partition table with hive as two fields. If the where condition does not specify all the two fields, an exception will occur Key: FLINK-20951 URL: https://issues.apache.org/jira/browse/FLINK-20951 Project: Flink Issue Type: Bug Components: Connectors / Hive, Table SQL / Runtime Affects Versions: 1.12.0 Environment: flink 1.12.0release-12 sql-cli Reporter: YUJIANBO The production hive table is partitioned by two fields：datekey and event I have do this test: （1）First: SELECT vid From table_A WHERE datekey = '20210112' AND event = 'XXX' AND vid = 'aa';(OK) SELECT vid From table_A WHERE datekey = '20210112' AND vid = 'aa'; (Error) （2）第二组对比 SELECT vid From table_B WHERE datekey = '20210112' AND event = 'YYY' AND vid = 'bb';(OK) SELECT vid From table_B WHERE datekey = '20210112' AND vid = 'bb'; (Error) The exception is: java.lang.RuntimeException: One or more fetchers have encountered exception at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:199) at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:154) at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:116) at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:273) at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:67) at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65) at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:395) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:191) at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:609) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:573) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570) at java.lang.Thread.run(Thread.java:745) Caused by: java.lang.RuntimeException: SplitFetcher thread 19 received unexpected exception while polling the records at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:146) at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:101) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ... 1 more Caused by: java.lang.IllegalArgumentException at java.nio.Buffer.position(Buffer.java:244) at org.apache.flink.hive.shaded.parquet.io.api.Binary$ByteBufferBackedBinary.getBytes(Binary.java:424) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.BytesColumnReader.readBatchFromDictionaryIds(BytesColumnReader.java:79) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.BytesColumnReader.readBatchFromDictionaryIds(BytesColumnReader.java:33) at org.apache.flink.hive.shaded.formats.parquet.vector.reader.AbstractColumnReader.readToVector(AbstractColumnReader.java:199) at org.apache.flink.hive.shaded.formats.parquet.ParquetVectorizedInputFormat$ParquetReader.nextBatch(ParquetVectorizedInputFormat.java:359) at org.apache.flink.hive.shaded.formats.parquet.ParquetVectorizedInputFormat$ParquetReader.readBatch(ParquetVectorizedInputFormat.java:328) at org.apache.flink.connector.file.src.impl.FileSourceSplitReader.fetch(FileSourceSplitReader.java:67) at org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:56) at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:138) ... 6 more Flink Email: http://apache-flink.147419.n8.nabble.com/Flink-sql-cli-hive-where-td10139.html -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #14606: [Flink-20876][table-planner-blink] Separate the implementation of StreamExecTemporalJoin
flinkbot edited a comment on pull request #14606: URL: https://github.com/apache/flink/pull/14606#issuecomment-757754275 ## CI report: * 3776b52cfe3535dcc193b3a922a7d1d658126d66 UNKNOWN * 155b18c169e45a97cd52c5b43883d5cf6b79f038 UNKNOWN * b6ca209812ae0e584d738f0ea6775a525a624a2f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11956) * 9276dee92aa80400b79a80168d19553aa2689a10 UNKNOWN Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] wuchong commented on a change in pull request #14604: [FLINK-20861][format/json] Provide an option for serializing DECIMALs in JSON as plain number instead of scientific notation
wuchong commented on a change in pull request #14604:
URL: https://github.com/apache/flink/pull/14604#discussion_r556296636
##
File path:
flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
##
@@ -562,6 +578,55 @@ public void testSerializationMapNullKey() throws Exception
{
assertEquals(expectResult3, new String(actual3));
}
+@Test
+public void testSerializationDecimalEncode() throws Exception {
+String original =
+
"{\"decimal1\":123.456789,\"decimal2\":454621864049246170,\"decimal3\":0.00027}";
+String expectedWithPlainDecimal =
+
"{\"decimal1\":123.456789,\"decimal2\":454621864049246170,\"decimal3\":0.00027}";
Review comment:
1. They are equal? We may only need to declare just one
`plainDecimalJson`?
2. Please move the variable near where it is used.
##
File path:
flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
##
@@ -562,6 +578,55 @@ public void testSerializationMapNullKey() throws Exception
{
assertEquals(expectResult3, new String(actual3));
}
+@Test
+public void testSerializationDecimalEncode() throws Exception {
+String original =
+
"{\"decimal1\":123.456789,\"decimal2\":454621864049246170,\"decimal3\":0.00027}";
+String expectedWithPlainDecimal =
+
"{\"decimal1\":123.456789,\"decimal2\":454621864049246170,\"decimal3\":0.00027}";
+String expectedWithoutPlainDecimal =
+
"{\"decimal1\":123.456789,\"decimal2\":4.5462186404924617E+17,\"decimal3\":2.7E-8}";
Review comment:
1. We can call this `scientificDecimalJson`.
2. Please move the variable near where it is used.
##
File path:
flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
##
@@ -562,6 +578,55 @@ public void testSerializationMapNullKey() throws Exception
{
assertEquals(expectResult3, new String(actual3));
}
+@Test
+public void testSerializationDecimalEncode() throws Exception {
+String original =
+
"{\"decimal1\":123.456789,\"decimal2\":454621864049246170,\"decimal3\":0.00027}";
+String expectedWithPlainDecimal =
+
"{\"decimal1\":123.456789,\"decimal2\":454621864049246170,\"decimal3\":0.00027}";
+String expectedWithoutPlainDecimal =
+
"{\"decimal1\":123.456789,\"decimal2\":4.5462186404924617E+17,\"decimal3\":2.7E-8}";
+
+RowType schema =
+(RowType)
+ROW(
+FIELD("decimal1", DECIMAL(9, 6)),
+FIELD("decimal2", DECIMAL(20, 0)),
+FIELD("decimal3", DECIMAL(11, 9)))
+.getLogicalType();
+
+TypeInformation resultTypeInfo = InternalTypeInfo.of(schema);
+
+JsonRowDataDeserializationSchema deserializationSchema =
+new JsonRowDataDeserializationSchema(
+schema, resultTypeInfo, false, false,
TimestampFormat.ISO_8601);
+
+JsonRowDataSerializationSchema serializationSchemaWithPlainDecimal =
+new JsonRowDataSerializationSchema(
+schema,
+TimestampFormat.ISO_8601,
+JsonOptions.MapNullKeyMode.LITERAL,
+"null",
+true);
+JsonRowDataSerializationSchema serializationSchemaWithoutPlainDecimal =
Review comment:
`scientificDecimalSerializer`
##
File path:
flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
##
@@ -562,6 +578,55 @@ public void testSerializationMapNullKey() throws Exception
{
assertEquals(expectResult3, new String(actual3));
}
+@Test
+public void testSerializationDecimalEncode() throws Exception {
+String original =
+
"{\"decimal1\":123.456789,\"decimal2\":454621864049246170,\"decimal3\":0.00027}";
+String expectedWithPlainDecimal =
+
"{\"decimal1\":123.456789,\"decimal2\":454621864049246170,\"decimal3\":0.00027}";
+String expectedWithoutPlainDecimal =
+
"{\"decimal1\":123.456789,\"decimal2\":4.5462186404924617E+17,\"decimal3\":2.7E-8}";
+
+RowType schema =
+(RowType)
+ROW(
+FIELD("decimal1", DECIMAL(9, 6)),
+FIELD("decimal2", DECIMAL(20, 0)),
+FIELD("decimal3", DECIMAL(11, 9)))
+.getLogicalType();
+
+TypeInformation resultTypeInfo =
[jira] [Updated] (FLINK-20936) Improvements in custom partition extractor
[
https://issues.apache.org/jira/browse/FLINK-20936?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
zhoukui updated FLINK-20936:
Description:
In flink1.12.0, I saw PartitionTimeExtractor is a constructor without
parameters, why not take the parameter “extractorPattern". See the source code
below:
{code:java}
@Experimental
public interface PartitionTimeExtractor extends Serializable {
String DEFAULT = "default";
String CUSTOM = "custom";
/**
* Extract time from partition keys and values.
*/
LocalDateTime extract(List partitionKeys, List
partitionValues);
static PartitionTimeExtractor create(
ClassLoader userClassLoader,
String extractorKind,
String extractorClass,
String extractorPattern) {
switch (extractorKind) {
case DEFAULT:
return new DefaultPartTimeExtractor(extractorPattern);
case CUSTOM:
try
{ Class aClass = userClassLoader.loadClass(extractorClass); return
(PartitionTimeExtractor) aClass.newInstance(); }
catch (ClassNotFoundException | IllegalAccessException | InstantiationException
e)
{ throw new RuntimeException( "Can not new instance for custom class from " +
extractorClass, e); }
default:
throw new UnsupportedOperationException(
"Unsupported extractor kind: " + extractorKind);
}
}
}
{code}
Would it be more reasonable to bring a parameter construction method? as
follows:
try
{ Class aClass = userClassLoader.loadClass(extractorClass); Constructor
declaredConstructor = aClass.getDeclaredConstructor(String.class);
return (PartitionTimeExtractor)
declaredConstructor.newInstance(extractorPattern); }
catch (ClassNotFoundException | IllegalAccessException | InstantiationException
| NoSuchMethodException | InvocationTargetException e)
{ throw new RuntimeException( "Can not new instance for custom class from " +
extractorClass, e); }
was:
In flink1.12.0, I saw PartitionTimeExtractor is a constructor without
parameters, why not take the parameter “extractorPattern". See the source code
below:
{code:java}
@Experimental
public interface PartitionTimeExtractor extends Serializable {
String DEFAULT = "default";
String CUSTOM = "custom";
/**
* Extract time from partition keys and values.
*/
LocalDateTime extract(List partitionKeys, List
partitionValues);
static PartitionTimeExtractor create(
ClassLoader userClassLoader,
String extractorKind,
String extractorClass,
String extractorPattern) {
switch (extractorKind) {
case DEFAULT:
return new DefaultPartTimeExtractor(extractorPattern);
case CUSTOM:
try
{ Class aClass = userClassLoader.loadClass(extractorClass); return
(PartitionTimeExtractor) aClass.newInstance(); }
catch (ClassNotFoundException | IllegalAccessException | InstantiationException
e)
{ throw new RuntimeException( "Can not new instance for custom class from " +
extractorClass, e); }
default:
throw new UnsupportedOperationException(
"Unsupported extractor kind: " + extractorKind);
}
}
}
{code}
Would it be more reasonable to bring a parameter construction method? as
follows:
try { Class aClass = userClassLoader.loadClass(extractorClass);
Constructor declaredConstructor =
aClass.getDeclaredConstructor(String.class); return (PartitionTimeExtractor)
declaredConstructor.newInstance(extractorPattern); } catch
(ClassNotFoundException | IllegalAccessException | InstantiationException |
NoSuchMethodException | InvocationTargetException e) { throw new
RuntimeException( "Can not new instance for custom class from " +
extractorClass, e); }
> Improvements in custom partition extractor
> ---
>
> Key: FLINK-20936
> URL: https://issues.apache.org/jira/browse/FLINK-20936
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / Hive
>Affects Versions: 1.12.0
> Environment: flink 1.12.0
> jdk 1.8
>Reporter: zhoukui
>Priority: Major
>
> In flink1.12.0, I saw PartitionTimeExtractor is a constructor without
> parameters, why not take the parameter “extractorPattern". See the source
> code below:
> {code:java}
> @Experimental
> public interface PartitionTimeExtractor extends Serializable {
> String DEFAULT = "default";
> String CUSTOM = "custom";
> /**
> * Extract time from partition keys and values.
> */
> LocalDateTime extract(List partitionKeys, List
> partitionValues);
> static PartitionTimeExtractor create(
> ClassLoader userClassLoader,
> String extractorKind,
> String extractorClass,
> String extractorPattern) {
> switch (extractorKind) {
> case DEFAULT:
> return new DefaultPartTimeExtractor(extractorPattern);
> case CUSTOM:
> try
> { Class aClass = userClassLoader.loadClass(extractorClass); return
> (PartitionTimeExtractor) aClass.newInstance(); }
> catch (ClassNotFoundException | IllegalAccessException |
> InstantiationException e)
> { throw new RuntimeException( "Can not new instance for
[jira] [Updated] (FLINK-20936) Improvements in custom partition extractor
[
https://issues.apache.org/jira/browse/FLINK-20936?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
zhoukui updated FLINK-20936:
Description:
In flink1.12.0, I saw PartitionTimeExtractor is a constructor without
parameters, why not take the parameter “extractorPattern". See the source code
below:
{code:java}
@Experimental
public interface PartitionTimeExtractor extends Serializable {
String DEFAULT = "default";
String CUSTOM = "custom";
/**
* Extract time from partition keys and values.
*/
LocalDateTime extract(List partitionKeys, List
partitionValues);
static PartitionTimeExtractor create(
ClassLoader userClassLoader,
String extractorKind,
String extractorClass,
String extractorPattern) {
switch (extractorKind) {
case DEFAULT:
return new DefaultPartTimeExtractor(extractorPattern);
case CUSTOM:
try
{ Class aClass = userClassLoader.loadClass(extractorClass); return
(PartitionTimeExtractor) aClass.newInstance(); }
catch (ClassNotFoundException | IllegalAccessException | InstantiationException
e)
{ throw new RuntimeException( "Can not new instance for custom class from " +
extractorClass, e); }
default:
throw new UnsupportedOperationException(
"Unsupported extractor kind: " + extractorKind);
}
}
}
{code}
Would it be more reasonable to bring a parameter construction method? as
follows:
try { Class aClass = userClassLoader.loadClass(extractorClass);
Constructor declaredConstructor =
aClass.getDeclaredConstructor(String.class); return (PartitionTimeExtractor)
declaredConstructor.newInstance(extractorPattern); } catch
(ClassNotFoundException | IllegalAccessException | InstantiationException |
NoSuchMethodException | InvocationTargetException e) { throw new
RuntimeException( "Can not new instance for custom class from " +
extractorClass, e); }
was:
In flink1.12.0, I saw PartitionTimeExtractor is a constructor without
parameters, why not take the parameter “extractorPattern". See the source code
below:
{code}
@Experimental
public interface PartitionTimeExtractor extends Serializable {
String DEFAULT = "default";
String CUSTOM = "custom";
/**
* Extract time from partition keys and values.
*/
LocalDateTime extract(List partitionKeys, List
partitionValues);
static PartitionTimeExtractor create(
ClassLoader userClassLoader,
String extractorKind,
String extractorClass,
String extractorPattern) {
switch (extractorKind) {
case DEFAULT:
return new DefaultPartTimeExtractor(extractorPattern);
case CUSTOM:
try
{ Class aClass = userClassLoader.loadClass(extractorClass); return
(PartitionTimeExtractor) aClass.newInstance(); }
catch (ClassNotFoundException | IllegalAccessException | InstantiationException
e)
{ throw new RuntimeException( "Can not new instance for custom class from " +
extractorClass, e); }
default:
throw new UnsupportedOperationException(
"Unsupported extractor kind: " + extractorKind);
}
}
}
Would it be more reasonable to bring a parameter construction method? as
follows:
try
{ Class aClass = userClassLoader.loadClass(extractorClass); Constructor
declaredConstructor = aClass.getDeclaredConstructor(String.class); return
(PartitionTimeExtractor) declaredConstructor.newInstance(extractorPattern); }
catch (ClassNotFoundException | IllegalAccessException | InstantiationException
| NoSuchMethodException | InvocationTargetException e)
{ throw new RuntimeException( "Can not new instance for custom class from " +
extractorClass, e); }
{code}
> Improvements in custom partition extractor
> ---
>
> Key: FLINK-20936
> URL: https://issues.apache.org/jira/browse/FLINK-20936
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / Hive
>Affects Versions: 1.12.0
> Environment: flink 1.12.0
> jdk 1.8
>Reporter: zhoukui
>Priority: Major
>
> In flink1.12.0, I saw PartitionTimeExtractor is a constructor without
> parameters, why not take the parameter “extractorPattern". See the source
> code below:
> {code:java}
> @Experimental
> public interface PartitionTimeExtractor extends Serializable {
> String DEFAULT = "default";
> String CUSTOM = "custom";
> /**
> * Extract time from partition keys and values.
> */
> LocalDateTime extract(List partitionKeys, List
> partitionValues);
> static PartitionTimeExtractor create(
> ClassLoader userClassLoader,
> String extractorKind,
> String extractorClass,
> String extractorPattern) {
> switch (extractorKind) {
> case DEFAULT:
> return new DefaultPartTimeExtractor(extractorPattern);
> case CUSTOM:
> try
> { Class aClass = userClassLoader.loadClass(extractorClass); return
> (PartitionTimeExtractor) aClass.newInstance(); }
> catch (ClassNotFoundException | IllegalAccessException |
> InstantiationException e)
> { throw new RuntimeException( "Can not new instance for
[jira] [Commented] (FLINK-20885) Exception when use 'canal-json.table.include' to filter Canal binlog but table contains 'source' column
[ https://issues.apache.org/jira/browse/FLINK-20885?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263917#comment-17263917 ] wangfei commented on FLINK-20885: - You can use the JSON above。 and canal-json.database.include and canal-json.table.include， It cannot be modified to support regular or wildcard characters？ > Exception when use 'canal-json.table.include' to filter Canal binlog but > table contains 'source' column > --- > > Key: FLINK-20885 > URL: https://issues.apache.org/jira/browse/FLINK-20885 > Project: Flink > Issue Type: Bug > Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table > SQL / Ecosystem >Reporter: Jark Wu >Assignee: Nicholas Jiang >Priority: Major > Attachments: image-2021-01-08-14-52-53-443.png, > image-2021-01-08-14-53-40-476.png, screenshot-1.png > > > I found a bug in the canal code. 'canal-json.table.include' does not filter > out the binlog of the specified table correctly, which will cause an error in > the parsing section. For example, if I want to read the binlog of > canal-json.table.include = 'a' table, there is a source field of int in table > a, but at this time if table b also has a source field of string, An error > will be reported. > !screenshot-1.png! -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-20885) Exception when use 'canal-json.table.include' to filter Canal binlog but table contains 'source' column
[
https://issues.apache.org/jira/browse/FLINK-20885?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263916#comment-17263916
]
wangfei commented on FLINK-20885:
-
|{"data":[\{"id":"0","username":"zhangsna","password":"passwd","AddTime":"2020-12-25
14:23:07"}],"database":"plink","es":1610520384000,"id":92151,"isDdl":false,"mysqlType":\{"id":"int(11)","username":"varchar(255)","password":"varchar(255)","AddTime":"datetime"},"old":[\{"password":""}],"pkNames":["id"],"sql":"","sqlType":\{"id":4,"username":12,"password":12,"AddTime":93},"table":"test_1","ts":1610520384579,"type":"UPDATE"}|
> Exception when use 'canal-json.table.include' to filter Canal binlog but
> table contains 'source' column
> ---
>
> Key: FLINK-20885
> URL: https://issues.apache.org/jira/browse/FLINK-20885
> Project: Flink
> Issue Type: Bug
> Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table
> SQL / Ecosystem
>Reporter: Jark Wu
>Assignee: Nicholas Jiang
>Priority: Major
> Attachments: image-2021-01-08-14-52-53-443.png,
> image-2021-01-08-14-53-40-476.png, screenshot-1.png
>
>
> I found a bug in the canal code. 'canal-json.table.include' does not filter
> out the binlog of the specified table correctly, which will cause an error in
> the parsing section. For example, if I want to read the binlog of
> canal-json.table.include = 'a' table, there is a source field of int in table
> a, but at this time if table b also has a source field of string, An error
> will be reported.
> !screenshot-1.png!
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] wuchong commented on a change in pull request #14604: [FLINK-20861][format/json] Provide an option for serializing DECIMALs in JSON as plain number instead of scientific notation
wuchong commented on a change in pull request #14604: URL: https://github.com/apache/flink/pull/14604#discussion_r556294884 ## File path: docs/dev/table/connectors/formats/json.md ## @@ -136,6 +136,13 @@ Format Options String Specify string literal to replace null key when 'json.map-null-key.mode' is LITERAL. + + json.encode.decimal-as-plain-number + optional + false + Boolean + Encode all decimals as plain numbers instead of possible scientific notations. e.g. 0.00027 is encoded as 2.7E-8 when this option is set to false. Review comment: Please also update this? 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: us...@infra.apache.org
[jira] [Comment Edited] (FLINK-20885) Exception when use 'canal-json.table.include' to filter Canal binlog but table contains 'source' column
[ https://issues.apache.org/jira/browse/FLINK-20885?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263913#comment-17263913 ] Nicholas Jiang edited comment on FLINK-20885 at 1/13/21, 6:43 AM: -- [~wangfeiair2324], the "data" field in binlog record you provided above has some problem to parse. Could you check the correct of the binlog record? And canal-json.database.include and canal-json.table.include only match one database and table. was (Author: nicholasjiang): [~wangfeiair2324], the "data" field in binlog record you provided above has some problem. Could you check the correct of the binlog record? And canal-json.database.include and canal-json.table.include only match one database and table. > Exception when use 'canal-json.table.include' to filter Canal binlog but > table contains 'source' column > --- > > Key: FLINK-20885 > URL: https://issues.apache.org/jira/browse/FLINK-20885 > Project: Flink > Issue Type: Bug > Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table > SQL / Ecosystem >Reporter: Jark Wu >Assignee: Nicholas Jiang >Priority: Major > Attachments: image-2021-01-08-14-52-53-443.png, > image-2021-01-08-14-53-40-476.png, screenshot-1.png > > > I found a bug in the canal code. 'canal-json.table.include' does not filter > out the binlog of the specified table correctly, which will cause an error in > the parsing section. For example, if I want to read the binlog of > canal-json.table.include = 'a' table, there is a source field of int in table > a, but at this time if table b also has a source field of string, An error > will be reported. > !screenshot-1.png! -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-20885) Exception when use 'canal-json.table.include' to filter Canal binlog but table contains 'source' column
[ https://issues.apache.org/jira/browse/FLINK-20885?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263913#comment-17263913 ] Nicholas Jiang commented on FLINK-20885: [~wangfeiair2324], the "data" field in binlog record you provided above has some problem. Could you check the correct of the binlog record? And canal-json.database.include and canal-json.table.include only match one database and table. > Exception when use 'canal-json.table.include' to filter Canal binlog but > table contains 'source' column > --- > > Key: FLINK-20885 > URL: https://issues.apache.org/jira/browse/FLINK-20885 > Project: Flink > Issue Type: Bug > Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table > SQL / Ecosystem >Reporter: Jark Wu >Assignee: Nicholas Jiang >Priority: Major > Attachments: image-2021-01-08-14-52-53-443.png, > image-2021-01-08-14-53-40-476.png, screenshot-1.png > > > I found a bug in the canal code. 'canal-json.table.include' does not filter > out the binlog of the specified table correctly, which will cause an error in > the parsing section. For example, if I want to read the binlog of > canal-json.table.include = 'a' table, there is a source field of int in table > a, but at this time if table b also has a source field of string, An error > will be reported. > !screenshot-1.png! -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-20950) SinkITCase.writerAndGlobalCommitterExecuteInStreamingMode test failed with "AssertionError"
[
https://issues.apache.org/jira/browse/FLINK-20950?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263912#comment-17263912
]
Huang Xingbo commented on FLINK-20950:
--
cc [~maguowei] Could you help take a look? Thanks.
> SinkITCase.writerAndGlobalCommitterExecuteInStreamingMode test failed with
> "AssertionError"
> ---
>
> Key: FLINK-20950
> URL: https://issues.apache.org/jira/browse/FLINK-20950
> Project: Flink
> Issue Type: Bug
> Components: API / DataStream
>Affects Versions: 1.13.0
>Reporter: Huang Xingbo
>Priority: Critical
> Labels: test-stability
>
> [https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=11940=logs=34f41360-6c0d-54d3-11a1-0292a2def1d9=2d56e022-1ace-542f-bf1a-b37dd63243f2]
> {code:java}
> 2021-01-12T16:47:00.7579536Z [ERROR] Failures:
> 2021-01-12T16:47:00.7580061Z [ERROR]
> SinkITCase.writerAndGlobalCommitterExecuteInStreamingMode:218
> 2021-01-12T16:47:00.7587186Z Expected: iterable over
> ["(895,null,-9223372036854775808)", "(895,null,-9223372036854775808)",
> "(127,null,-9223372036854775808)", "(127,null,-9223372036854775808)",
> "(148,null,-9223372036854775808)", "(148,null,-9223372036854775808)",
> "(161,null,-9223372036854775808)", "(161,null,-9223372036854775808)",
> "(148,null,-9223372036854775808)", "(148,null,-9223372036854775808)",
> "(662,null,-9223372036854775808)", "(662,null,-9223372036854775808)",
> "(822,null,-9223372036854775808)", "(822,null,-9223372036854775808)",
> "(491,null,-9223372036854775808)", "(491,null,-9223372036854775808)",
> "(275,null,-9223372036854775808)", "(275,null,-9223372036854775808)",
> "(122,null,-9223372036854775808)", "(122,null,-9223372036854775808)",
> "(850,null,-9223372036854775808)", "(850,null,-9223372036854775808)",
> "(630,null,-9223372036854775808)", "(630,null,-9223372036854775808)",
> "(682,null,-9223372036854775808)", "(682,null,-9223372036854775808)",
> "(765,null,-9223372036854775808)", "(765,null,-9223372036854775808)",
> "(434,null,-9223372036854775808)", "(434,null,-9223372036854775808)",
> "(970,null,-9223372036854775808)", "(970,null,-9223372036854775808)",
> "(714,null,-9223372036854775808)", "(714,null,-9223372036854775808)",
> "(795,null,-9223372036854775808)", "(795,null,-9223372036854775808)",
> "(288,null,-9223372036854775808)", "(288,null,-9223372036854775808)",
> "(422,null,-9223372036854775808)", "(422,null,-9223372036854775808)"] in any
> order
> 2021-01-12T16:47:00.7591663Z but: Not matched: "end of input"
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Created] (FLINK-20950) SinkITCase.writerAndGlobalCommitterExecuteInStreamingMode test failed with "AssertionError"
Huang Xingbo created FLINK-20950:
Summary: SinkITCase.writerAndGlobalCommitterExecuteInStreamingMode
test failed with "AssertionError"
Key: FLINK-20950
URL: https://issues.apache.org/jira/browse/FLINK-20950
Project: Flink
Issue Type: Bug
Components: API / DataStream
Affects Versions: 1.13.0
Reporter: Huang Xingbo
[https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=11940=logs=34f41360-6c0d-54d3-11a1-0292a2def1d9=2d56e022-1ace-542f-bf1a-b37dd63243f2]
{code:java}
2021-01-12T16:47:00.7579536Z [ERROR] Failures:
2021-01-12T16:47:00.7580061Z [ERROR]
SinkITCase.writerAndGlobalCommitterExecuteInStreamingMode:218
2021-01-12T16:47:00.7587186Z Expected: iterable over
["(895,null,-9223372036854775808)", "(895,null,-9223372036854775808)",
"(127,null,-9223372036854775808)", "(127,null,-9223372036854775808)",
"(148,null,-9223372036854775808)", "(148,null,-9223372036854775808)",
"(161,null,-9223372036854775808)", "(161,null,-9223372036854775808)",
"(148,null,-9223372036854775808)", "(148,null,-9223372036854775808)",
"(662,null,-9223372036854775808)", "(662,null,-9223372036854775808)",
"(822,null,-9223372036854775808)", "(822,null,-9223372036854775808)",
"(491,null,-9223372036854775808)", "(491,null,-9223372036854775808)",
"(275,null,-9223372036854775808)", "(275,null,-9223372036854775808)",
"(122,null,-9223372036854775808)", "(122,null,-9223372036854775808)",
"(850,null,-9223372036854775808)", "(850,null,-9223372036854775808)",
"(630,null,-9223372036854775808)", "(630,null,-9223372036854775808)",
"(682,null,-9223372036854775808)", "(682,null,-9223372036854775808)",
"(765,null,-9223372036854775808)", "(765,null,-9223372036854775808)",
"(434,null,-9223372036854775808)", "(434,null,-9223372036854775808)",
"(970,null,-9223372036854775808)", "(970,null,-9223372036854775808)",
"(714,null,-9223372036854775808)", "(714,null,-9223372036854775808)",
"(795,null,-9223372036854775808)", "(795,null,-9223372036854775808)",
"(288,null,-9223372036854775808)", "(288,null,-9223372036854775808)",
"(422,null,-9223372036854775808)", "(422,null,-9223372036854775808)"] in any
order
2021-01-12T16:47:00.7591663Z but: Not matched: "end of input"
{code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #14606: [Flink-20876][table-planner-blink] Separate the implementation of StreamExecTemporalJoin
flinkbot edited a comment on pull request #14606: URL: https://github.com/apache/flink/pull/14606#issuecomment-757754275 ## CI report: * c0b4f7fc8369909215cc174a341aa70068b3c0da Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11951) * 3776b52cfe3535dcc193b3a922a7d1d658126d66 UNKNOWN * 155b18c169e45a97cd52c5b43883d5cf6b79f038 UNKNOWN * b6ca209812ae0e584d738f0ea6775a525a624a2f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11956) * 9276dee92aa80400b79a80168d19553aa2689a10 UNKNOWN Bot commands 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 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: us...@infra.apache.org
[jira] [Commented] (FLINK-20926) Allow to read metadata for maxwell-json format
[ https://issues.apache.org/jira/browse/FLINK-20926?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263910#comment-17263910 ] Q Kang commented on FLINK-20926: [~jark] `primary-key-columns` maps to `primary_key_columns` field in Maxwell, which has exactly the same meaning as `pk-names` in Canal. Should we use `pk-names` instead or just stick to `primary-key-columns`? As for the extra fields, I agree with you that we just leave them for now. > Allow to read metadata for maxwell-json format > -- > > Key: FLINK-20926 > URL: https://issues.apache.org/jira/browse/FLINK-20926 > Project: Flink > Issue Type: Sub-task > Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table > SQL / API >Reporter: Q Kang >Assignee: Q Kang >Priority: Major > > Canal-json and debezium-json metadata are already implemented. We still need > metadata for maxwell-json to fill in the blanks for CDC formats. > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] KarmaGYZ commented on a change in pull request #14612: [FLINK-20864][runtime] Introduce the DEFAULT resource profile for the…
KarmaGYZ commented on a change in pull request #14612: URL: https://github.com/apache/flink/pull/14612#discussion_r556287199 ## File path: flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTableImpl.java ## @@ -319,7 +321,7 @@ public boolean allocateSlot( taskSlot = new TaskSlot<>( index, -resourceProfile, +effectiveResourceProfile, Review comment: Yes, it should actually be a hotfix for FLINK-20837. We treat the matching of resource profile as one of the conditions to judge the duplicate slot. So we need to move this transition forward. 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: us...@infra.apache.org
[jira] [Updated] (FLINK-20947) Unable to trigger Hive partition commit when use table sql to write data from Kafka into a Hive table
[
https://issues.apache.org/jira/browse/FLINK-20947?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu updated FLINK-20947:
Fix Version/s: 1.12.1
> Unable to trigger Hive partition commit when use table sql to write data from
> Kafka into a Hive table
> -
>
> Key: FLINK-20947
> URL: https://issues.apache.org/jira/browse/FLINK-20947
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.12.0
>Reporter: Weijia Xu
>Assignee: Shengkai Fang
>Priority: Major
> Fix For: 1.13.0, 1.12.1
>
>
> I use table sql to create stream with kafka source, and write data from Kafka
> into a Hive partitioned table.
> The following sql to create kafka table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE stream_tmp.kafka_tmp (`messageId` STRING, `message_type`
> STRING,`payload` STRING,`timestamp` BIGINT, " +
> " procTime AS PROCTIME()," +
> " eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp` /
> 1000,'-MM-dd HH:mm:ss'))," +
> " WATERMARK FOR eventTime AS eventTime - INTERVAL '15' SECOND )"
> +
> " WITH ('connector' = 'kafka'," +
> " 'topic' = 'XXX-topic'," +
> " 'properties.bootstrap.servers'='kafka-server:9092'," +
> " 'properties.group.id' = 'XXX-group_id'," +
> " 'scan.startup.mode' = 'latest-offset'," +
> " 'format' = 'json'," +
> " 'json.fail-on-missing-field' = 'false'," +
> " 'json.ignore-parse-errors' = 'true' )"
> );{code}
>
> And, the following sql to create Hive table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE hive_tmp.kafka_hive_tmp (`message_id`
> STRING,`message_type` STRING,`payload` STRING,`event_ts` BIGINT ) " +
> " PARTITIONED BY (ts_date STRING,ts_hour STRING, ts_minute
> STRING)" +
> " STORED AS PARQUET TBLPROPERTIES (" +
> " 'sink.partition-commit.trigger' = 'partition-time'," +
> " 'sink.partition-commit.delay' = '1 min'," +
> " 'sink.partition-commit.policy.kind'='metastore,success-file',"
> +
> " 'sink.partition-commit.success-file.name'='_SUCCESS'," +
> " 'partition.time-extractor.timestamp-pattern' = '$ts_date
> $ts_hour:$ts_minute:00')");
> {code}
>
> For the kafka topic used above, which has multi partitions, with some of
> the partitions there's data coming in, while other partitions with no data
> coming.
> I noticed that there's config "_table.exec.source.idle-timeout_" can handle
> the situation for the "idle" source partition, but event though set this
> config, it still cannot trigger the Hive partition commit (that means the
> "_SUCCESS" file will not be created for the partition).
> After do the analysis for this issue, find the root cause is that the
> watermark for the "idle" partition will not advance, which cause the Hive
> partition cannot be committed.
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (FLINK-20947) Unable to trigger Hive partition commit when use table sql to write data from Kafka into a Hive table
[
https://issues.apache.org/jira/browse/FLINK-20947?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu updated FLINK-20947:
Fix Version/s: (was: 1.12.1)
1.12.2
> Unable to trigger Hive partition commit when use table sql to write data from
> Kafka into a Hive table
> -
>
> Key: FLINK-20947
> URL: https://issues.apache.org/jira/browse/FLINK-20947
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.12.0
>Reporter: Weijia Xu
>Assignee: Shengkai Fang
>Priority: Major
> Fix For: 1.13.0, 1.12.2
>
>
> I use table sql to create stream with kafka source, and write data from Kafka
> into a Hive partitioned table.
> The following sql to create kafka table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE stream_tmp.kafka_tmp (`messageId` STRING, `message_type`
> STRING,`payload` STRING,`timestamp` BIGINT, " +
> " procTime AS PROCTIME()," +
> " eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp` /
> 1000,'-MM-dd HH:mm:ss'))," +
> " WATERMARK FOR eventTime AS eventTime - INTERVAL '15' SECOND )"
> +
> " WITH ('connector' = 'kafka'," +
> " 'topic' = 'XXX-topic'," +
> " 'properties.bootstrap.servers'='kafka-server:9092'," +
> " 'properties.group.id' = 'XXX-group_id'," +
> " 'scan.startup.mode' = 'latest-offset'," +
> " 'format' = 'json'," +
> " 'json.fail-on-missing-field' = 'false'," +
> " 'json.ignore-parse-errors' = 'true' )"
> );{code}
>
> And, the following sql to create Hive table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE hive_tmp.kafka_hive_tmp (`message_id`
> STRING,`message_type` STRING,`payload` STRING,`event_ts` BIGINT ) " +
> " PARTITIONED BY (ts_date STRING,ts_hour STRING, ts_minute
> STRING)" +
> " STORED AS PARQUET TBLPROPERTIES (" +
> " 'sink.partition-commit.trigger' = 'partition-time'," +
> " 'sink.partition-commit.delay' = '1 min'," +
> " 'sink.partition-commit.policy.kind'='metastore,success-file',"
> +
> " 'sink.partition-commit.success-file.name'='_SUCCESS'," +
> " 'partition.time-extractor.timestamp-pattern' = '$ts_date
> $ts_hour:$ts_minute:00')");
> {code}
>
> For the kafka topic used above, which has multi partitions, with some of
> the partitions there's data coming in, while other partitions with no data
> coming.
> I noticed that there's config "_table.exec.source.idle-timeout_" can handle
> the situation for the "idle" source partition, but event though set this
> config, it still cannot trigger the Hive partition commit (that means the
> "_SUCCESS" file will not be created for the partition).
> After do the analysis for this issue, find the root cause is that the
> watermark for the "idle" partition will not advance, which cause the Hive
> partition cannot be committed.
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Comment Edited] (FLINK-20947) Unable to trigger Hive partition commit when use table sql to write data from Kafka into a Hive table
[
https://issues.apache.org/jira/browse/FLINK-20947?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263907#comment-17263907
]
Jark Wu edited comment on FLINK-20947 at 1/13/21, 6:20 AM:
---
Good catch [~weijiaxu]! I think you are right.
[~fsk119] will help to fix this. We should think about how to add a good test
for this. It is not test covered when supporting the watermark pushdown.
was (Author: jark):
Good catch [~weijiaxu]! I think you are right.
> Unable to trigger Hive partition commit when use table sql to write data from
> Kafka into a Hive table
> -
>
> Key: FLINK-20947
> URL: https://issues.apache.org/jira/browse/FLINK-20947
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.12.0
>Reporter: Weijia Xu
>Assignee: Shengkai Fang
>Priority: Major
> Fix For: 1.13.0
>
>
> I use table sql to create stream with kafka source, and write data from Kafka
> into a Hive partitioned table.
> The following sql to create kafka table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE stream_tmp.kafka_tmp (`messageId` STRING, `message_type`
> STRING,`payload` STRING,`timestamp` BIGINT, " +
> " procTime AS PROCTIME()," +
> " eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp` /
> 1000,'-MM-dd HH:mm:ss'))," +
> " WATERMARK FOR eventTime AS eventTime - INTERVAL '15' SECOND )"
> +
> " WITH ('connector' = 'kafka'," +
> " 'topic' = 'XXX-topic'," +
> " 'properties.bootstrap.servers'='kafka-server:9092'," +
> " 'properties.group.id' = 'XXX-group_id'," +
> " 'scan.startup.mode' = 'latest-offset'," +
> " 'format' = 'json'," +
> " 'json.fail-on-missing-field' = 'false'," +
> " 'json.ignore-parse-errors' = 'true' )"
> );{code}
>
> And, the following sql to create Hive table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE hive_tmp.kafka_hive_tmp (`message_id`
> STRING,`message_type` STRING,`payload` STRING,`event_ts` BIGINT ) " +
> " PARTITIONED BY (ts_date STRING,ts_hour STRING, ts_minute
> STRING)" +
> " STORED AS PARQUET TBLPROPERTIES (" +
> " 'sink.partition-commit.trigger' = 'partition-time'," +
> " 'sink.partition-commit.delay' = '1 min'," +
> " 'sink.partition-commit.policy.kind'='metastore,success-file',"
> +
> " 'sink.partition-commit.success-file.name'='_SUCCESS'," +
> " 'partition.time-extractor.timestamp-pattern' = '$ts_date
> $ts_hour:$ts_minute:00')");
> {code}
>
> For the kafka topic used above, which has multi partitions, with some of
> the partitions there's data coming in, while other partitions with no data
> coming.
> I noticed that there's config "_table.exec.source.idle-timeout_" can handle
> the situation for the "idle" source partition, but event though set this
> config, it still cannot trigger the Hive partition commit (that means the
> "_SUCCESS" file will not be created for the partition).
> After do the analysis for this issue, find the root cause is that the
> watermark for the "idle" partition will not advance, which cause the Hive
> partition cannot be committed.
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-20495) Elasticsearch6DynamicSinkITCase Hang
[
https://issues.apache.org/jira/browse/FLINK-20495?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263908#comment-17263908
]
Huang Xingbo commented on FLINK-20495:
--
[https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=11949=logs=d44f43ce-542c-597d-bf94-b0718c71e5e8=03dca39c-73e8-5aaf-601d-328ae5c35f20]
> Elasticsearch6DynamicSinkITCase Hang
>
>
> Key: FLINK-20495
> URL: https://issues.apache.org/jira/browse/FLINK-20495
> Project: Flink
> Issue Type: Bug
> Components: Connectors / ElasticSearch
>Affects Versions: 1.13.0
>Reporter: Huang Xingbo
>Priority: Major
> Labels: test-stability
>
> [https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=10535=logs=d44f43ce-542c-597d-bf94-b0718c71e5e8=03dca39c-73e8-5aaf-601d-328ae5c35f20]
>
> {code:java}
> 2020-12-04T22:39:33.9748225Z [INFO] Running
> org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch6DynamicSinkITCase
> 2020-12-04T22:54:51.9486410Z
> ==
> 2020-12-04T22:54:51.9488766Z Process produced no output for 900 seconds.
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-20947) Unable to trigger Hive partition commit when use table sql to write data from Kafka into a Hive table
[
https://issues.apache.org/jira/browse/FLINK-20947?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263907#comment-17263907
]
Jark Wu commented on FLINK-20947:
-
Good catch [~weijiaxu]! I think you are right.
> Unable to trigger Hive partition commit when use table sql to write data from
> Kafka into a Hive table
> -
>
> Key: FLINK-20947
> URL: https://issues.apache.org/jira/browse/FLINK-20947
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.12.0
>Reporter: Weijia Xu
>Assignee: Weijia Xu
>Priority: Major
> Fix For: 1.13.0
>
>
> I use table sql to create stream with kafka source, and write data from Kafka
> into a Hive partitioned table.
> The following sql to create kafka table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE stream_tmp.kafka_tmp (`messageId` STRING, `message_type`
> STRING,`payload` STRING,`timestamp` BIGINT, " +
> " procTime AS PROCTIME()," +
> " eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp` /
> 1000,'-MM-dd HH:mm:ss'))," +
> " WATERMARK FOR eventTime AS eventTime - INTERVAL '15' SECOND )"
> +
> " WITH ('connector' = 'kafka'," +
> " 'topic' = 'XXX-topic'," +
> " 'properties.bootstrap.servers'='kafka-server:9092'," +
> " 'properties.group.id' = 'XXX-group_id'," +
> " 'scan.startup.mode' = 'latest-offset'," +
> " 'format' = 'json'," +
> " 'json.fail-on-missing-field' = 'false'," +
> " 'json.ignore-parse-errors' = 'true' )"
> );{code}
>
> And, the following sql to create Hive table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE hive_tmp.kafka_hive_tmp (`message_id`
> STRING,`message_type` STRING,`payload` STRING,`event_ts` BIGINT ) " +
> " PARTITIONED BY (ts_date STRING,ts_hour STRING, ts_minute
> STRING)" +
> " STORED AS PARQUET TBLPROPERTIES (" +
> " 'sink.partition-commit.trigger' = 'partition-time'," +
> " 'sink.partition-commit.delay' = '1 min'," +
> " 'sink.partition-commit.policy.kind'='metastore,success-file',"
> +
> " 'sink.partition-commit.success-file.name'='_SUCCESS'," +
> " 'partition.time-extractor.timestamp-pattern' = '$ts_date
> $ts_hour:$ts_minute:00')");
> {code}
>
> For the kafka topic used above, which has multi partitions, with some of
> the partitions there's data coming in, while other partitions with no data
> coming.
> I noticed that there's config "_table.exec.source.idle-timeout_" can handle
> the situation for the "idle" source partition, but event though set this
> config, it still cannot trigger the Hive partition commit (that means the
> "_SUCCESS" file will not be created for the partition).
> After do the analysis for this issue, find the root cause is that the
> watermark for the "idle" partition will not advance, which cause the Hive
> partition cannot be committed.
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Assigned] (FLINK-20947) Unable to trigger Hive partition commit when use table sql to write data from Kafka into a Hive table
[
https://issues.apache.org/jira/browse/FLINK-20947?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu reassigned FLINK-20947:
---
Assignee: Shengkai Fang (was: Weijia Xu)
> Unable to trigger Hive partition commit when use table sql to write data from
> Kafka into a Hive table
> -
>
> Key: FLINK-20947
> URL: https://issues.apache.org/jira/browse/FLINK-20947
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.12.0
>Reporter: Weijia Xu
>Assignee: Shengkai Fang
>Priority: Major
> Fix For: 1.13.0
>
>
> I use table sql to create stream with kafka source, and write data from Kafka
> into a Hive partitioned table.
> The following sql to create kafka table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE stream_tmp.kafka_tmp (`messageId` STRING, `message_type`
> STRING,`payload` STRING,`timestamp` BIGINT, " +
> " procTime AS PROCTIME()," +
> " eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp` /
> 1000,'-MM-dd HH:mm:ss'))," +
> " WATERMARK FOR eventTime AS eventTime - INTERVAL '15' SECOND )"
> +
> " WITH ('connector' = 'kafka'," +
> " 'topic' = 'XXX-topic'," +
> " 'properties.bootstrap.servers'='kafka-server:9092'," +
> " 'properties.group.id' = 'XXX-group_id'," +
> " 'scan.startup.mode' = 'latest-offset'," +
> " 'format' = 'json'," +
> " 'json.fail-on-missing-field' = 'false'," +
> " 'json.ignore-parse-errors' = 'true' )"
> );{code}
>
> And, the following sql to create Hive table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE hive_tmp.kafka_hive_tmp (`message_id`
> STRING,`message_type` STRING,`payload` STRING,`event_ts` BIGINT ) " +
> " PARTITIONED BY (ts_date STRING,ts_hour STRING, ts_minute
> STRING)" +
> " STORED AS PARQUET TBLPROPERTIES (" +
> " 'sink.partition-commit.trigger' = 'partition-time'," +
> " 'sink.partition-commit.delay' = '1 min'," +
> " 'sink.partition-commit.policy.kind'='metastore,success-file',"
> +
> " 'sink.partition-commit.success-file.name'='_SUCCESS'," +
> " 'partition.time-extractor.timestamp-pattern' = '$ts_date
> $ts_hour:$ts_minute:00')");
> {code}
>
> For the kafka topic used above, which has multi partitions, with some of
> the partitions there's data coming in, while other partitions with no data
> coming.
> I noticed that there's config "_table.exec.source.idle-timeout_" can handle
> the situation for the "idle" source partition, but event though set this
> config, it still cannot trigger the Hive partition commit (that means the
> "_SUCCESS" file will not be created for the partition).
> After do the analysis for this issue, find the root cause is that the
> watermark for the "idle" partition will not advance, which cause the Hive
> partition cannot be committed.
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] lmagic233 commented on a change in pull request #14604: [FLINK-20861][format/json] Provide an option for serializing DECIMALs in JSON as plain number instead of scientific notation
lmagic233 commented on a change in pull request #14604:
URL: https://github.com/apache/flink/pull/14604#discussion_r556286139
##
File path:
flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
##
@@ -562,6 +579,56 @@ public void testSerializationMapNullKey() throws Exception
{
assertEquals(expectResult3, new String(actual3));
}
+@Test
+public void testSerializationDecimalEncode() throws Exception {
+BigDecimal decimal1 = new BigDecimal("123.456789");
+BigDecimal decimal2 = new BigDecimal("454621864049246170");
+BigDecimal decimal3 = new BigDecimal("0.00027");
+
+ObjectMapper objectMapper = new ObjectMapper();
+
objectMapper.configure(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN, true);
+
+ObjectNode root = objectMapper.createObjectNode();
+root.put("decimal1", decimal1);
+root.put("decimal2", decimal2);
+root.put("decimal3", decimal3);
+
+byte[] serializedJson = objectMapper.writeValueAsBytes(root);
+
+DataType dataType =
+ROW(
+FIELD("decimal1", DECIMAL(9, 6)),
+FIELD("decimal2", DECIMAL(20, 0)),
+FIELD("decimal3", DECIMAL(11, 9)));
+
+RowType schema = (RowType) dataType.getLogicalType();
+TypeInformation resultTypeInfo = InternalTypeInfo.of(schema);
+
+JsonRowDataDeserializationSchema deserializationSchema =
+new JsonRowDataDeserializationSchema(
+schema, resultTypeInfo, false, false,
TimestampFormat.ISO_8601);
+
+Row expected = new Row(3);
+expected.setField(0, decimal1);
+expected.setField(1, decimal2);
+expected.setField(2, decimal3);
+
+RowData rowData = deserializationSchema.deserialize(serializedJson);
Review comment:
Ditto.
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:
us...@infra.apache.org
[jira] [Assigned] (FLINK-20947) Unable to trigger Hive partition commit when use table sql to write data from Kafka into a Hive table
[
https://issues.apache.org/jira/browse/FLINK-20947?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu reassigned FLINK-20947:
---
Assignee: Weijia Xu
> Unable to trigger Hive partition commit when use table sql to write data from
> Kafka into a Hive table
> -
>
> Key: FLINK-20947
> URL: https://issues.apache.org/jira/browse/FLINK-20947
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.12.0
>Reporter: Weijia Xu
>Assignee: Weijia Xu
>Priority: Major
> Fix For: 1.13.0
>
>
> I use table sql to create stream with kafka source, and write data from Kafka
> into a Hive partitioned table.
> The following sql to create kafka table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE stream_tmp.kafka_tmp (`messageId` STRING, `message_type`
> STRING,`payload` STRING,`timestamp` BIGINT, " +
> " procTime AS PROCTIME()," +
> " eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp` /
> 1000,'-MM-dd HH:mm:ss'))," +
> " WATERMARK FOR eventTime AS eventTime - INTERVAL '15' SECOND )"
> +
> " WITH ('connector' = 'kafka'," +
> " 'topic' = 'XXX-topic'," +
> " 'properties.bootstrap.servers'='kafka-server:9092'," +
> " 'properties.group.id' = 'XXX-group_id'," +
> " 'scan.startup.mode' = 'latest-offset'," +
> " 'format' = 'json'," +
> " 'json.fail-on-missing-field' = 'false'," +
> " 'json.ignore-parse-errors' = 'true' )"
> );{code}
>
> And, the following sql to create Hive table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE hive_tmp.kafka_hive_tmp (`message_id`
> STRING,`message_type` STRING,`payload` STRING,`event_ts` BIGINT ) " +
> " PARTITIONED BY (ts_date STRING,ts_hour STRING, ts_minute
> STRING)" +
> " STORED AS PARQUET TBLPROPERTIES (" +
> " 'sink.partition-commit.trigger' = 'partition-time'," +
> " 'sink.partition-commit.delay' = '1 min'," +
> " 'sink.partition-commit.policy.kind'='metastore,success-file',"
> +
> " 'sink.partition-commit.success-file.name'='_SUCCESS'," +
> " 'partition.time-extractor.timestamp-pattern' = '$ts_date
> $ts_hour:$ts_minute:00')");
> {code}
>
> For the kafka topic used above, which has multi partitions, with some of
> the partitions there's data coming in, while other partitions with no data
> coming.
> I noticed that there's config "_table.exec.source.idle-timeout_" can handle
> the situation for the "idle" source partition, but event though set this
> config, it still cannot trigger the Hive partition commit (that means the
> "_SUCCESS" file will not be created for the partition).
> After do the analysis for this issue, find the root cause is that the
> watermark for the "idle" partition will not advance, which cause the Hive
> partition cannot be committed.
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] lmagic233 commented on a change in pull request #14604: [FLINK-20861][format/json] Provide an option for serializing DECIMALs in JSON as plain number instead of scientific notation
lmagic233 commented on a change in pull request #14604:
URL: https://github.com/apache/flink/pull/14604#discussion_r556286068
##
File path:
flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowDataSerDeSchemaTest.java
##
@@ -562,6 +579,56 @@ public void testSerializationMapNullKey() throws Exception
{
assertEquals(expectResult3, new String(actual3));
}
+@Test
+public void testSerializationDecimalEncode() throws Exception {
+BigDecimal decimal1 = new BigDecimal("123.456789");
+BigDecimal decimal2 = new BigDecimal("454621864049246170");
+BigDecimal decimal3 = new BigDecimal("0.00027");
+
+ObjectMapper objectMapper = new ObjectMapper();
+
objectMapper.configure(JsonGenerator.Feature.WRITE_BIGDECIMAL_AS_PLAIN, true);
+
+ObjectNode root = objectMapper.createObjectNode();
+root.put("decimal1", decimal1);
+root.put("decimal2", decimal2);
+root.put("decimal3", decimal3);
+
+byte[] serializedJson = objectMapper.writeValueAsBytes(root);
Review comment:
Modified test case to use explicit JSON strings.
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:
us...@infra.apache.org
[GitHub] [flink] wenlong88 commented on a change in pull request #14605: [FLINK-20883][table-planner-blink] Separate the implementation of BatchExecOverAggregate and StreamExecOverAggregate
wenlong88 commented on a change in pull request #14605:
URL: https://github.com/apache/flink/pull/14605#discussion_r556283456
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/over/MultiFieldRangeBoundComparatorCodeGenerator.scala
##
@@ -48,9 +46,14 @@ class MultiFieldRangeBoundComparatorCodeGenerator(
}
val ctx = CodeGeneratorContext(conf)
-
val compareCode = GenerateUtils.generateRowCompare(
Review comment:
use the new method accepts sortSpec
##
File path:
flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecOverAggregate.java
##
@@ -0,0 +1,380 @@
+/*
+ * 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.planner.plan.nodes.exec.stream;
+
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
+import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
+import org.apache.flink.table.planner.codegen.agg.AggsHandlerCodeGenerator;
+import org.apache.flink.table.planner.delegation.PlannerBase;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
+import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase;
+import org.apache.flink.table.planner.plan.nodes.exec.utils.OverSpec;
+import org.apache.flink.table.planner.plan.utils.AggregateInfoList;
+import org.apache.flink.table.planner.plan.utils.AggregateUtil;
+import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
+import org.apache.flink.table.planner.plan.utils.OverAggregateUtil;
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil;
+import org.apache.flink.table.runtime.generated.GeneratedAggsHandleFunction;
+import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
+import
org.apache.flink.table.runtime.operators.over.ProcTimeRangeBoundedPrecedingFunction;
+import
org.apache.flink.table.runtime.operators.over.ProcTimeRowsBoundedPrecedingFunction;
+import
org.apache.flink.table.runtime.operators.over.ProcTimeUnboundedPrecedingFunction;
+import
org.apache.flink.table.runtime.operators.over.RowTimeRangeBoundedPrecedingFunction;
+import
org.apache.flink.table.runtime.operators.over.RowTimeRangeUnboundedPrecedingFunction;
+import
org.apache.flink.table.runtime.operators.over.RowTimeRowsBoundedPrecedingFunction;
+import
org.apache.flink.table.runtime.operators.over.RowTimeRowsUnboundedPrecedingFunction;
+import org.apache.flink.table.runtime.types.LogicalTypeDataTypeConverter;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.TimestampKind;
+import org.apache.flink.table.types.logical.TimestampType;
+
+import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.tools.RelBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.IntStream;
+
+/** Stream {@link ExecNode} for time-based over operator. */
+public class StreamExecOverAggregate extends ExecNodeBase
+implements StreamExecNode {
+private static final Logger LOG =
LoggerFactory.getLogger(StreamExecOverAggregate.class);
+
+private final OverSpec overSpec;
+
+public StreamExecOverAggregate(
+OverSpec overSpec, ExecEdge inputEdge, RowType outputType, String
description) {
+super(Collections.singletonList(inputEdge), outputType, description);
+
[GitHub] [flink] KarmaGYZ commented on a change in pull request #14612: [FLINK-20864][runtime] Introduce the DEFAULT resource profile for the…
KarmaGYZ commented on a change in pull request #14612:
URL: https://github.com/apache/flink/pull/14612#discussion_r556283807
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceProfile.java
##
@@ -271,13 +271,41 @@ public boolean isMatching(final ResourceProfile required)
{
return true;
}
-if (cpuCores.getValue().compareTo(required.cpuCores.getValue()) >= 0
-&& taskHeapMemory.compareTo(required.taskHeapMemory) >= 0
-&& taskOffHeapMemory.compareTo(required.taskOffHeapMemory) >= 0
-&& managedMemory.compareTo(required.managedMemory) >= 0
-&& networkMemory.compareTo(required.networkMemory) >= 0) {
+return false;
+}
+
+/**
+ * Check whether this resource profile is bigger than the given resource
profile.
+ *
+ * @param other the other resource profile
+ * @return true if this resource profile is bigger, otherwise false
+ */
+public boolean isBiggerThan(final ResourceProfile other) {
Review comment:
`allFieldsNoLessThan` sounds good to me.
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:
us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14625: [FLINK-20941][table-planner-blink] Introduce StreamPhysicalMatch, and make StreamExecMatch only extended from ExecNode
flinkbot edited a comment on pull request #14625: URL: https://github.com/apache/flink/pull/14625#issuecomment-759222713 ## CI report: * fdf9c3005c53149eb7be9501c3c451c4e5cb77e5 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11963) Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #14625: [FLINK-20941][table-planner-blink] Introduce StreamPhysicalMatch, and make StreamExecMatch only extended from ExecNode
flinkbot commented on pull request #14625: URL: https://github.com/apache/flink/pull/14625#issuecomment-759222713 ## CI report: * fdf9c3005c53149eb7be9501c3c451c4e5cb77e5 UNKNOWN Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14604: [FLINK-20861][format/json] Provide an option for serializing DECIMALs in JSON as plain number instead of scientific notation
flinkbot edited a comment on pull request #14604: URL: https://github.com/apache/flink/pull/14604#issuecomment-757521857 ## CI report: * 6f16ecba1741139347662f1a892cca2885b603c9 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11862) * 34a633e801bcb8100e0962b358b4e400a47c7047 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11959) * 095aa26dba7a04800cb203e252774d0d7655cc25 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11962) Bot commands 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 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: us...@infra.apache.org
[jira] [Assigned] (FLINK-20948) Separate the implementation of StreamExecDeduplicate
[ https://issues.apache.org/jira/browse/FLINK-20948?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] godfrey he reassigned FLINK-20948: -- Assignee: godfrey he > Separate the implementation of StreamExecDeduplicate > > > Key: FLINK-20948 > URL: https://issues.apache.org/jira/browse/FLINK-20948 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Reporter: godfrey he >Assignee: godfrey he >Priority: Major > Fix For: 1.13.0 > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-20949) Separate the implementation of sink nodes
godfrey he created FLINK-20949: -- Summary: Separate the implementation of sink nodes Key: FLINK-20949 URL: https://issues.apache.org/jira/browse/FLINK-20949 Project: Flink Issue Type: Sub-task Components: Table SQL / Planner Reporter: godfrey he Fix For: 1.13.0 -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-20948) Separate the implementation of StreamExecDeduplicate
godfrey he created FLINK-20948: -- Summary: Separate the implementation of StreamExecDeduplicate Key: FLINK-20948 URL: https://issues.apache.org/jira/browse/FLINK-20948 Project: Flink Issue Type: Sub-task Components: Table SQL / Planner Reporter: godfrey he Fix For: 1.13.0 -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot commented on pull request #14625: [FLINK-20941][table-planner-blink] Introduce StreamPhysicalMatch, and make StreamExecMatch only extended from ExecNode
flinkbot commented on pull request #14625: URL: https://github.com/apache/flink/pull/14625#issuecomment-759218410 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 fdf9c3005c53149eb7be9501c3c451c4e5cb77e5 (Wed Jan 13 05:39:49 UTC 2021) **Warnings:** * No documentation files were touched! Remember to keep the Flink docs up to date! Mention the bot in a comment to re-run the automated checks. ## 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. 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 Bot commands 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 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: us...@infra.apache.org
[jira] [Commented] (FLINK-20751) Port batch python group aggregate nodes to Java
[ https://issues.apache.org/jira/browse/FLINK-20751?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263896#comment-17263896 ] godfrey he commented on FLINK-20751: cc [~hxbks2ks] > Port batch python group aggregate nodes to Java > --- > > Key: FLINK-20751 > URL: https://issues.apache.org/jira/browse/FLINK-20751 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Reporter: godfrey he >Priority: Major > Fix For: 1.13.0 > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (FLINK-20941) Separate the implementation of StreamExecMatch
[ https://issues.apache.org/jira/browse/FLINK-20941?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-20941: --- Labels: pull-request-available (was: ) > Separate the implementation of StreamExecMatch > -- > > Key: FLINK-20941 > URL: https://issues.apache.org/jira/browse/FLINK-20941 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Reporter: godfrey he >Assignee: godfrey he >Priority: Major > Labels: pull-request-available > Fix For: 1.13.0 > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] godfreyhe opened a new pull request #14625: [FLINK-20941][table-planner-blink] Introduce StreamPhysicalMatch, and make StreamExecMatch only extended from ExecNode
godfreyhe opened a new pull request #14625: URL: https://github.com/apache/flink/pull/14625 ## What is the purpose of the change *Separate the implementation of StreamExecMatch* ## Brief change log - *Introduce StreamPhysicalMatch, and make StreamExecMatch only extended from ExecNode* ## Verifying this change This change is a refactoring rework covered by existing tests. ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): (yes / **no**) - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**) - The serializers: (yes / **no** / don't know) - The runtime per-record code paths (performance sensitive): (yes / **no** / don't know) - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (yes / **no** / don't know) - The S3 file system connector: (yes / **no** / don't know) ## Documentation - Does this pull request introduce a new feature? (yes / **no**) - If yes, how is the feature documented? (not applicable / docs / JavaDocs / **not documented**) 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: us...@infra.apache.org
[jira] [Commented] (FLINK-20947) Unable to trigger Hive partition commit when use table sql to write data from Kafka into a Hive table
[
https://issues.apache.org/jira/browse/FLINK-20947?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263895#comment-17263895
]
Weijia Xu commented on FLINK-20947:
---
I think the root cause is the watermark for the "idle" partition will not
advance, the reason why this happens, is because when set the
"_table.exec.source.idle-timeout_", and the idle timeout value is valid, the
watermark strategy that enriched with idleness detection should be updated, but
actually not (after checking from the source code)
source code class: _PushWatermarkIntoTableSourceScanRuleBase.java_
> Unable to trigger Hive partition commit when use table sql to write data from
> Kafka into a Hive table
> -
>
> Key: FLINK-20947
> URL: https://issues.apache.org/jira/browse/FLINK-20947
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.12.0
>Reporter: Weijia Xu
>Priority: Major
> Fix For: 1.13.0
>
>
> I use table sql to create stream with kafka source, and write data from Kafka
> into a Hive partitioned table.
> The following sql to create kafka table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE stream_tmp.kafka_tmp (`messageId` STRING, `message_type`
> STRING,`payload` STRING,`timestamp` BIGINT, " +
> " procTime AS PROCTIME()," +
> " eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp` /
> 1000,'-MM-dd HH:mm:ss'))," +
> " WATERMARK FOR eventTime AS eventTime - INTERVAL '15' SECOND )"
> +
> " WITH ('connector' = 'kafka'," +
> " 'topic' = 'XXX-topic'," +
> " 'properties.bootstrap.servers'='kafka-server:9092'," +
> " 'properties.group.id' = 'XXX-group_id'," +
> " 'scan.startup.mode' = 'latest-offset'," +
> " 'format' = 'json'," +
> " 'json.fail-on-missing-field' = 'false'," +
> " 'json.ignore-parse-errors' = 'true' )"
> );{code}
>
> And, the following sql to create Hive table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE hive_tmp.kafka_hive_tmp (`message_id`
> STRING,`message_type` STRING,`payload` STRING,`event_ts` BIGINT ) " +
> " PARTITIONED BY (ts_date STRING,ts_hour STRING, ts_minute
> STRING)" +
> " STORED AS PARQUET TBLPROPERTIES (" +
> " 'sink.partition-commit.trigger' = 'partition-time'," +
> " 'sink.partition-commit.delay' = '1 min'," +
> " 'sink.partition-commit.policy.kind'='metastore,success-file',"
> +
> " 'sink.partition-commit.success-file.name'='_SUCCESS'," +
> " 'partition.time-extractor.timestamp-pattern' = '$ts_date
> $ts_hour:$ts_minute:00')");
> {code}
>
> For the kafka topic used above, which has multi partitions, with some of
> the partitions there's data coming in, while other partitions with no data
> coming.
> I noticed that there's config "_table.exec.source.idle-timeout_" can handle
> the situation for the "idle" source partition, but event though set this
> config, it still cannot trigger the Hive partition commit (that means the
> "_SUCCESS" file will not be created for the partition).
> After do the analysis for this issue, find the root cause is that the
> watermark for the "idle" partition will not advance, which cause the Hive
> partition cannot be committed.
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #14620: [FLINK-20940][table-planner] Use session time zone in LOCALTIME/LOCALTIMSTAMP functions
flinkbot edited a comment on pull request #14620: URL: https://github.com/apache/flink/pull/14620#issuecomment-758696085 ## CI report: * 9476884989f58f1f3ecc2bf33e9d225524d3d0b1 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11941) * a23b91048435a844d5c7b8acb174a3424c78a68a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11961) Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14620: [FLINK-20940][table-planner] Use session time zone in LOCALTIME/LOCALTIMSTAMP functions
flinkbot edited a comment on pull request #14620: URL: https://github.com/apache/flink/pull/14620#issuecomment-758696085 ## CI report: * 9476884989f58f1f3ecc2bf33e9d225524d3d0b1 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11941) * a23b91048435a844d5c7b8acb174a3424c78a68a UNKNOWN Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14617: [hotfix][doc] Fix default format in document for temporal function FROM_UNIXTIME(numeric, format)
flinkbot edited a comment on pull request #14617: URL: https://github.com/apache/flink/pull/14617#issuecomment-758573504 ## CI report: * 6fda825ac66b34b6e55cb89ad24ea962928e1ef1 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11960) Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14606: [Flink-20876][table-planner-blink] Separate the implementation of StreamExecTemporalJoin
flinkbot edited a comment on pull request #14606: URL: https://github.com/apache/flink/pull/14606#issuecomment-757754275 ## CI report: * c0b4f7fc8369909215cc174a341aa70068b3c0da Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11951) * 3776b52cfe3535dcc193b3a922a7d1d658126d66 UNKNOWN * 155b18c169e45a97cd52c5b43883d5cf6b79f038 UNKNOWN * b6ca209812ae0e584d738f0ea6775a525a624a2f Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11956) Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14604: [FLINK-20861][format/json] Provide an option for serializing DECIMALs in JSON as plain number instead of scientific notation
flinkbot edited a comment on pull request #14604: URL: https://github.com/apache/flink/pull/14604#issuecomment-757521857 ## CI report: * 6f16ecba1741139347662f1a892cca2885b603c9 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11862) * 34a633e801bcb8100e0962b358b4e400a47c7047 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11959) * 095aa26dba7a04800cb203e252774d0d7655cc25 UNKNOWN Bot commands 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 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: us...@infra.apache.org
[jira] [Updated] (FLINK-20947) Unable to trigger Hive partition commit when use table sql to write data from Kafka into a Hive table
[
https://issues.apache.org/jira/browse/FLINK-20947?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jark Wu updated FLINK-20947:
Fix Version/s: 1.13.0
> Unable to trigger Hive partition commit when use table sql to write data from
> Kafka into a Hive table
> -
>
> Key: FLINK-20947
> URL: https://issues.apache.org/jira/browse/FLINK-20947
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.12.0
>Reporter: Weijia Xu
>Priority: Major
> Fix For: 1.13.0
>
>
> I use table sql to create stream with kafka source, and write data from Kafka
> into a Hive partitioned table.
> The following sql to create kafka table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE stream_tmp.kafka_tmp (`messageId` STRING, `message_type`
> STRING,`payload` STRING,`timestamp` BIGINT, " +
> " procTime AS PROCTIME()," +
> " eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp` /
> 1000,'-MM-dd HH:mm:ss'))," +
> " WATERMARK FOR eventTime AS eventTime - INTERVAL '15' SECOND )"
> +
> " WITH ('connector' = 'kafka'," +
> " 'topic' = 'XXX-topic'," +
> " 'properties.bootstrap.servers'='kafka-server:9092'," +
> " 'properties.group.id' = 'XXX-group_id'," +
> " 'scan.startup.mode' = 'latest-offset'," +
> " 'format' = 'json'," +
> " 'json.fail-on-missing-field' = 'false'," +
> " 'json.ignore-parse-errors' = 'true' )"
> );{code}
>
> And, the following sql to create Hive table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE hive_tmp.kafka_hive_tmp (`message_id`
> STRING,`message_type` STRING,`payload` STRING,`event_ts` BIGINT ) " +
> " PARTITIONED BY (ts_date STRING,ts_hour STRING, ts_minute
> STRING)" +
> " STORED AS PARQUET TBLPROPERTIES (" +
> " 'sink.partition-commit.trigger' = 'partition-time'," +
> " 'sink.partition-commit.delay' = '1 min'," +
> " 'sink.partition-commit.policy.kind'='metastore,success-file',"
> +
> " 'sink.partition-commit.success-file.name'='_SUCCESS'," +
> " 'partition.time-extractor.timestamp-pattern' = '$ts_date
> $ts_hour:$ts_minute:00')");
> {code}
>
> For the kafka topic used above, which has multi partitions, with some of
> the partitions there's data coming in, while other partitions with no data
> coming.
> I noticed that there's config "_table.exec.source.idle-timeout_" can handle
> the situation for the "idle" source partition, but event though set this
> config, it still cannot trigger the Hive partition commit (that means the
> "_SUCCESS" file will not be created for the partition).
> After do the analysis for this issue, find the root cause is that the
> watermark for the "idle" partition will not advance, which cause the Hive
> partition cannot be committed.
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] xintongsong commented on a change in pull request #14612: [FLINK-20864][runtime] Introduce the DEFAULT resource profile for the…
xintongsong commented on a change in pull request #14612:
URL: https://github.com/apache/flink/pull/14612#discussion_r556238763
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceProfile.java
##
@@ -271,13 +271,41 @@ public boolean isMatching(final ResourceProfile required)
{
return true;
}
-if (cpuCores.getValue().compareTo(required.cpuCores.getValue()) >= 0
-&& taskHeapMemory.compareTo(required.taskHeapMemory) >= 0
-&& taskOffHeapMemory.compareTo(required.taskOffHeapMemory) >= 0
-&& managedMemory.compareTo(required.managedMemory) >= 0
-&& networkMemory.compareTo(required.networkMemory) >= 0) {
+return false;
+}
Review comment:
I wonder if we can remove the `if (this.equals(UNKNOWN))` branch now,
and add `throwUnsupportedOperationExecptionIfUnknown` at the beginning of this
method.
Ideally, the caller of this method should always represent a resource
capacity rather than a requirement, thus should never be `UNKNOWN`. I suspect
the reason we didn't do that previously is because this method was also used as
`isBiggerThan` in some test cases.
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/slot/TaskSlotTableImpl.java
##
@@ -319,7 +321,7 @@ public boolean allocateSlot(
taskSlot =
new TaskSlot<>(
index,
-resourceProfile,
+effectiveResourceProfile,
Review comment:
I think changes to this file are less related to the matching rules.
Basically, for static slot requests, we now use the default profile rather
than `UNKNOWN` for creating the `TaskSlot`.
* Could you explain why is this change needed?
* Even if this change is indeed needed, we should make it a separate commit.
##
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/types/ResourceProfileTest.java
##
@@ -102,23 +102,61 @@ public void testMatchRequirement() {
.setManagedMemoryMB(100)
.setNetworkMemoryMB(100)
.build();
-assertFalse(rp4.isMatching(rp5));
+assertFalse(rp4.isBiggerThan(rp5));
ResourceSpec rs1 = ResourceSpec.newBuilder(1.0,
100).setGPUResource(2.2).build();
ResourceSpec rs2 = ResourceSpec.newBuilder(1.0,
100).setGPUResource(1.1).build();
-assertFalse(rp1.isMatching(ResourceProfile.fromResourceSpec(rs1)));
+assertFalse(rp1.isBiggerThan(ResourceProfile.fromResourceSpec(rs1)));
assertTrue(
ResourceProfile.fromResourceSpec(rs1)
-.isMatching(ResourceProfile.fromResourceSpec(rs2)));
+.isBiggerThan(ResourceProfile.fromResourceSpec(rs2)));
assertFalse(
ResourceProfile.fromResourceSpec(rs2)
-.isMatching(ResourceProfile.fromResourceSpec(rs1)));
+.isBiggerThan(ResourceProfile.fromResourceSpec(rs1)));
}
@Test
-public void testUnknownMatchesUnknown() {
Review comment:
Why is this test case removed? I think we still need to cover `UNKNOWN
>= UNKNOWN`.
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceProfile.java
##
@@ -271,13 +271,41 @@ public boolean isMatching(final ResourceProfile required)
{
return true;
}
-if (cpuCores.getValue().compareTo(required.cpuCores.getValue()) >= 0
-&& taskHeapMemory.compareTo(required.taskHeapMemory) >= 0
-&& taskOffHeapMemory.compareTo(required.taskOffHeapMemory) >= 0
-&& managedMemory.compareTo(required.managedMemory) >= 0
-&& networkMemory.compareTo(required.networkMemory) >= 0) {
+return false;
+}
+
+/**
+ * Check whether this resource profile is bigger than the given resource
profile.
+ *
+ * @param other the other resource profile
+ * @return true if this resource profile is bigger, otherwise false
+ */
+public boolean isBiggerThan(final ResourceProfile other) {
Review comment:
I'm not sure about the name `isBiggerThan`.
* First of all, the semantic here is `>=` rather than `>`
* I'm not sure `bigger` is the proper phrase for describing resource amount.
Maybe `greater` or `larger`.
* I think we should emphasize the semantic that **every** field should be
`>= other`, to avoid confusing this with comparing the **total** resource.
To sum up, I would suggest `allFieldsGreaterThanOrEqualTo`,
`allFieldsLargerThanOrEqualTo`, or `allFieldsNoLessThan`.
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceProfile.java
##
@@ -271,13
[jira] [Commented] (FLINK-20947) Unable to trigger Hive partition commit when use table sql to write data from Kafka into a Hive table
[
https://issues.apache.org/jira/browse/FLINK-20947?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263885#comment-17263885
]
Jark Wu commented on FLINK-20947:
-
cc [~fsk119] could you help to check whether the idle timeout works for the
push downed watermark?
[~weijiaxu] could you share the explain result of your query?
> Unable to trigger Hive partition commit when use table sql to write data from
> Kafka into a Hive table
> -
>
> Key: FLINK-20947
> URL: https://issues.apache.org/jira/browse/FLINK-20947
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.12.0
>Reporter: Weijia Xu
>Priority: Major
>
> I use table sql to create stream with kafka source, and write data from Kafka
> into a Hive partitioned table.
> The following sql to create kafka table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE stream_tmp.kafka_tmp (`messageId` STRING, `message_type`
> STRING,`payload` STRING,`timestamp` BIGINT, " +
> " procTime AS PROCTIME()," +
> " eventTime AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp` /
> 1000,'-MM-dd HH:mm:ss'))," +
> " WATERMARK FOR eventTime AS eventTime - INTERVAL '15' SECOND )"
> +
> " WITH ('connector' = 'kafka'," +
> " 'topic' = 'XXX-topic'," +
> " 'properties.bootstrap.servers'='kafka-server:9092'," +
> " 'properties.group.id' = 'XXX-group_id'," +
> " 'scan.startup.mode' = 'latest-offset'," +
> " 'format' = 'json'," +
> " 'json.fail-on-missing-field' = 'false'," +
> " 'json.ignore-parse-errors' = 'true' )"
> );{code}
>
> And, the following sql to create Hive table:
> {code:java}
> // code placeholder
> tableEnv.executeSql(
> "CREATE TABLE hive_tmp.kafka_hive_tmp (`message_id`
> STRING,`message_type` STRING,`payload` STRING,`event_ts` BIGINT ) " +
> " PARTITIONED BY (ts_date STRING,ts_hour STRING, ts_minute
> STRING)" +
> " STORED AS PARQUET TBLPROPERTIES (" +
> " 'sink.partition-commit.trigger' = 'partition-time'," +
> " 'sink.partition-commit.delay' = '1 min'," +
> " 'sink.partition-commit.policy.kind'='metastore,success-file',"
> +
> " 'sink.partition-commit.success-file.name'='_SUCCESS'," +
> " 'partition.time-extractor.timestamp-pattern' = '$ts_date
> $ts_hour:$ts_minute:00')");
> {code}
>
> For the kafka topic used above, which has multi partitions, with some of
> the partitions there's data coming in, while other partitions with no data
> coming.
> I noticed that there's config "_table.exec.source.idle-timeout_" can handle
> the situation for the "idle" source partition, but event though set this
> config, it still cannot trigger the Hive partition commit (that means the
> "_SUCCESS" file will not be created for the partition).
> After do the analysis for this issue, find the root cause is that the
> watermark for the "idle" partition will not advance, which cause the Hive
> partition cannot be committed.
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on pull request #14624: [FLINK-20750][python][table-planner-blink] Port stream python group aggregate nodes to Java
flinkbot edited a comment on pull request #14624: URL: https://github.com/apache/flink/pull/14624#issuecomment-759191873 ## CI report: * 062f813ecf60bcc973829ab9e2ace0bdf1afa969 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11958) Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14617: [hotfix][doc] Fix default format in document for temporal function FROM_UNIXTIME(numeric, format)
flinkbot edited a comment on pull request #14617: URL: https://github.com/apache/flink/pull/14617#issuecomment-758573504 ## CI report: * 090ba49bfd2cc3b452a70d64b1a907e599195e8d Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11922) * 6fda825ac66b34b6e55cb89ad24ea962928e1ef1 UNKNOWN Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14606: [Flink-20876][table-planner-blink] Separate the implementation of StreamExecTemporalJoin
flinkbot edited a comment on pull request #14606: URL: https://github.com/apache/flink/pull/14606#issuecomment-757754275 ## CI report: * 34b5cc968f647c2a7c2e9fdd24c36de5f86f2624 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11937) * c0b4f7fc8369909215cc174a341aa70068b3c0da Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11951) * 3776b52cfe3535dcc193b3a922a7d1d658126d66 UNKNOWN * 155b18c169e45a97cd52c5b43883d5cf6b79f038 UNKNOWN * b6ca209812ae0e584d738f0ea6775a525a624a2f Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11956) Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14604: [FLINK-20861][format/json] Provide an option for serializing DECIMALs in JSON as plain number instead of scientific notation
flinkbot edited a comment on pull request #14604: URL: https://github.com/apache/flink/pull/14604#issuecomment-757521857 ## CI report: * 6f16ecba1741139347662f1a892cca2885b603c9 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11862) * 34a633e801bcb8100e0962b358b4e400a47c7047 UNKNOWN Bot commands 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 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: us...@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #14530: [FLINK-20348][kafka] Make "schema-registry.subject" optional for Kafka sink with avro-confluent format
flinkbot edited a comment on pull request #14530: URL: https://github.com/apache/flink/pull/14530#issuecomment-752828495 ## CI report: * 35003647ea219b5b67317e240a0dcd85860dbe51 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11919) * 0ff0b34168bf6262b65de24d71c335f80d8ef91a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=11957) Bot commands 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 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: us...@infra.apache.org
[jira] [Commented] (FLINK-20932) sql client.blink streaming to make it checkpoint enabled [HOw |supported now?]
[ https://issues.apache.org/jira/browse/FLINK-20932?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17263881#comment-17263881 ] Jark Wu commented on FLINK-20932: - [~nhe150] yes, you are right. This is the one enables checkpoint. > sql client.blink streaming to make it checkpoint enabled [HOw |supported now?] > -- > > Key: FLINK-20932 > URL: https://issues.apache.org/jira/browse/FLINK-20932 > Project: Flink > Issue Type: Bug >Reporter: Norman He >Priority: Major > Attachments: Screen Shot 2021-01-11 at 3.54.46 PM.png, > flink-conf.yaml, sql.yaml > > > no example how to support streaming checkpoints in flink embedded sql client. -- This message was sent by Atlassian Jira (v8.3.4#803005)
