You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by "kuansix (via GitHub)" <gi...@apache.org> on 2023/08/08 06:57:33 UTC
[GitHub] [inlong] kuansix opened a new pull request, #8664: [INLONG-7908][Sort] Postgres connector supports parallel read
kuansix opened a new pull request, #8664:
URL: https://github.com/apache/inlong/pull/8664
### Prepare a Pull Request
*(Change the title refer to the following example)*
- [INLONG-7908][Sort] Postgres connector supports parallel read
- Fixes #7908
### Motivation
*Explain here the context, and why you're making that change. What is the problem you're trying to solve?*
### Modifications
*Describe the modifications you've done.*
### Verifying this change
*(Please pick either of the following options)*
- [ ] This change is a trivial rework/code cleanup without any test coverage.
- [ ] This change is already covered by existing tests, such as:
*(please describe tests)*
- [ ] This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
### Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
- If a feature is not applicable for documentation, explain why?
- If a feature is not documented yet in this PR, please create a follow-up issue for adding the documentation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] e-mhui commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "e-mhui (via GitHub)" <gi...@apache.org>.
e-mhui commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1322317712
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/source/config/PostgresSourceConfig.java:
##########
@@ -100,7 +101,7 @@ public RelationalTableFilters getTableFilters() {
@Override
public List<String> getMetricLabelList() {
- return Arrays.asList(AbstractSourceInfo.DATABASE_NAME_KEY,
+ return Arrays.asList(DATABASE_NAME_KEY,
Review Comment:
`cdc-base` module already has `DATABASE_NAME` constant, so you can import it, not define it by yourself.
<img width="696" alt="image" src="https://github.com/apache/inlong/assets/111486498/5492f9c2-5b83-459a-9ddb-32e39e74cdf2">
you also refer Oracle CDC.
<img width="1040" alt="image" src="https://github.com/apache/inlong/assets/111486498/feaa5bec-9a8e-4614-a262-2fe7cc0fb9f5">
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] EMsnap commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "EMsnap (via GitHub)" <gi...@apache.org>.
EMsnap commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1292938656
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/io/debezium/connector/postgresql/PostgresObjectFactory.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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 io.debezium.connector.postgresql;
+
+import io.debezium.connector.postgresql.connection.PostgresConnection;
+import io.debezium.connector.postgresql.connection.ReplicationConnection;
+import io.debezium.relational.TableId;
+import io.debezium.schema.TopicSelector;
+import io.debezium.util.Clock;
+import io.debezium.util.Metronome;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.time.ZoneOffset;
+
+/**
+ * A factory for creating various Debezium objects
+ *
+ * <p>It is a hack to access package-private constructor in debezium.
Review Comment:
Please add license for these files
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] EMsnap commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "EMsnap (via GitHub)" <gi...@apache.org>.
EMsnap commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1294130255
##########
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/config/JdbcSourceConfig.java:
##########
@@ -49,6 +50,7 @@ public abstract class JdbcSourceConfig extends BaseSourceConfig {
public JdbcSourceConfig(
StartupOptions startupOptions,
List<String> databaseList,
+ List<String> schemaList,
Review Comment:
schemaList is never used
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] EMsnap commented on pull request #8664: [INLONG-7908][Sort] Postgres connector supports parallel read
Posted by "EMsnap (via GitHub)" <gi...@apache.org>.
EMsnap commented on PR #8664:
URL: https://github.com/apache/inlong/pull/8664#issuecomment-1670554371
Please check the ut and compile thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] dockerzhang merged pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "dockerzhang (via GitHub)" <gi...@apache.org>.
dockerzhang merged PR #8664:
URL: https://github.com/apache/inlong/pull/8664
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] e-mhui commented on a diff in pull request #8664: [INLONG-7908][Sort] Postgres connector supports parallel read
Posted by "e-mhui (via GitHub)" <gi...@apache.org>.
e-mhui commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1286684644
##########
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/debezium/table/RowDataDebeziumDeserializeSchema.java:
##########
@@ -757,11 +757,6 @@ private void extractDdlRecord(SourceRecord record, Collector<RowData> out, Table
}
- @Override
Review Comment:
It should not be deleted.
##########
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/IncrementalSource.java:
##########
@@ -128,7 +128,7 @@ public IncrementalSourceReader<T, C> createReader(SourceReaderContext readerCont
.withRegisterMetric(RegisteredMetric.ALL)
.build();
- sourceReaderMetrics.registerMetrics(metricOption, metricConfig.getMetricLabelList());
Review Comment:
It should not be deleted.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] kuansix commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "kuansix (via GitHub)" <gi...@apache.org>.
kuansix commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1324428028
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/config/OracleSourceConfigFactory.java:
##########
@@ -108,10 +108,6 @@ public OracleSourceConfig create(int subtaskId) {
props.setProperty("database.port", String.valueOf(port));
}
- if (schemaList != null) {
Review Comment:
schemaList is never used for pg and oracle
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] EMsnap commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "EMsnap (via GitHub)" <gi...@apache.org>.
EMsnap commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1294131902
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/source/options/PostgresSourceOptions.java:
##########
@@ -0,0 +1,83 @@
+/*
+ * 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.inlong.sort.cdc.postgres.source.options;
+
+import org.apache.inlong.sort.cdc.postgres.source.PostgresSourceBuilder;
+
+import com.ververica.cdc.connectors.base.options.JdbcSourceOptions;
+import com.ververica.cdc.debezium.table.DebeziumChangelogMode;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+import java.time.Duration;
+
+/** Configurations for {@link PostgresSourceBuilder.PostgresIncrementalSource}. */
+public class PostgresSourceOptions extends JdbcSourceOptions {
+
+ public static final ConfigOption<Integer> PORT =
+ ConfigOptions.key("port")
+ .intType()
+ .defaultValue(5432)
+ .withDescription("Integer port number of the PostgreSQL database server.");
+
+ public static final ConfigOption<String> DECODING_PLUGIN_NAME =
Review Comment:
The config is never used
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] gong commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "gong (via GitHub)" <gi...@apache.org>.
gong commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1325781999
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/source/reader/PostgresSourceRecordEmitter.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.inlong.sort.cdc.postgres.source.reader;
+
+import org.apache.inlong.sort.cdc.base.debezium.DebeziumDeserializationSchema;
+import org.apache.inlong.sort.cdc.base.debezium.history.FlinkJsonTableChangeSerializer;
+import org.apache.inlong.sort.cdc.base.source.meta.offset.Offset;
+import org.apache.inlong.sort.cdc.base.source.meta.offset.OffsetFactory;
+import org.apache.inlong.sort.cdc.base.source.meta.split.SourceSplitState;
+import org.apache.inlong.sort.cdc.base.source.metrics.SourceReaderMetrics;
+import org.apache.inlong.sort.cdc.base.source.reader.IncrementalSourceReader;
+import org.apache.inlong.sort.cdc.base.source.reader.IncrementalSourceRecordEmitter;
+import org.apache.inlong.sort.cdc.base.util.RecordUtils;
+
+import io.debezium.connector.AbstractSourceInfo;
+import io.debezium.data.Envelope;
+import io.debezium.document.Array;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.HistoryRecord;
+import io.debezium.relational.history.TableChanges;
+import io.debezium.relational.history.TableChanges.TableChange;
+import org.apache.flink.api.connector.source.SourceOutput;
+import org.apache.flink.connector.base.source.reader.RecordEmitter;
+import org.apache.flink.util.Collector;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+import static com.ververica.cdc.connectors.base.source.meta.wartermark.WatermarkEvent.isHighWatermarkEvent;
+import static com.ververica.cdc.connectors.base.source.meta.wartermark.WatermarkEvent.isWatermarkEvent;
+import static com.ververica.cdc.connectors.base.utils.SourceRecordUtils.getHistoryRecord;
+import static com.ververica.cdc.connectors.base.utils.SourceRecordUtils.isDataChangeRecord;
+import static com.ververica.cdc.connectors.base.utils.SourceRecordUtils.isSchemaChangeEvent;
Review Comment:
Fix dependency to `org.apache.inlong.sort.cdc`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] kuansix commented on pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "kuansix (via GitHub)" <gi...@apache.org>.
kuansix commented on PR #8664:
URL: https://github.com/apache/inlong/pull/8664#issuecomment-1676619142
> Please check the ut and compile thanks
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] kuansix commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "kuansix (via GitHub)" <gi...@apache.org>.
kuansix commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1324430702
##########
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/IncrementalSourceReader.java:
##########
@@ -237,7 +242,7 @@ private StreamSplit discoverTableSchemasForStreamSplit(StreamSplit split) {
public void handleSourceEvents(SourceEvent sourceEvent) {
if (sourceEvent instanceof FinishedSnapshotSplitsAckEvent) {
FinishedSnapshotSplitsAckEvent ackEvent = (FinishedSnapshotSplitsAckEvent) sourceEvent;
- LOG.debug(
+ LOG.info(
Review Comment:
finished snapshot splits only once, and is necessary when debug online.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] kuansix commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "kuansix (via GitHub)" <gi...@apache.org>.
kuansix commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1324428028
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/config/OracleSourceConfigFactory.java:
##########
@@ -108,10 +108,6 @@ public OracleSourceConfig create(int subtaskId) {
props.setProperty("database.port", String.valueOf(port));
}
- if (schemaList != null) {
Review Comment:
schemaList is never used for pg and oracle
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] dockerzhang commented on pull request #8664: [INLONG-7908][Sort] Postgres connector supports parallel read
Posted by "dockerzhang (via GitHub)" <gi...@apache.org>.
dockerzhang commented on PR #8664:
URL: https://github.com/apache/inlong/pull/8664#issuecomment-1669039766
@kuansix, please add more description for this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] gong commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "gong (via GitHub)" <gi...@apache.org>.
gong commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1307208277
##########
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/meta/offset/OffsetDeserializerSerializer.java:
##########
@@ -84,7 +84,8 @@ default OffsetDeserializer createOffsetDeserializer() {
default FinishedSnapshotSplitInfo deserialize(byte[] serialized) {
try {
final DataInputDeserializer in = new DataInputDeserializer(serialized);
- TableId tableId = TableId.parse(in.readUTF());
+ boolean useCatalogBeforeSchema = in.readBoolean();
+ TableId tableId = TableId.parse(in.readUTF(), useCatalogBeforeSchema);
Review Comment:
It should improve refer to https://github.com/ververica/flink-cdc-connectors/pull/2216
```java
default FinishedSnapshotSplitInfo deserialize(byte[] serialized) {
try {
final DataInputDeserializer in = new DataInputDeserializer(serialized);
String tableIdStr = in.readUTF();
String splitId = in.readUTF();
Object[] splitStart = serializedStringToRow(in.readUTF());
Object[] splitEnd = serializedStringToRow(in.readUTF());
OffsetFactory offsetFactory = (OffsetFactory) serializedStringToObject(in.readUTF());
Offset highWatermark = readOffsetPosition(in);
boolean useCatalogBeforeSchema = true;
if (in.available() > 0) {
useCatalogBeforeSchema = in.readBoolean();
}
TableId tableId = TableId.parse(tableIdStr, useCatalogBeforeSchema);
in.releaseArrays();
return new FinishedSnapshotSplitInfo(
tableId, splitId, splitStart, splitEnd, highWatermark, offsetFactory);
} catch (IOException e) {
throw new FlinkRuntimeException(e);
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] gong commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "gong (via GitHub)" <gi...@apache.org>.
gong commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1325781999
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/source/reader/PostgresSourceRecordEmitter.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.inlong.sort.cdc.postgres.source.reader;
+
+import org.apache.inlong.sort.cdc.base.debezium.DebeziumDeserializationSchema;
+import org.apache.inlong.sort.cdc.base.debezium.history.FlinkJsonTableChangeSerializer;
+import org.apache.inlong.sort.cdc.base.source.meta.offset.Offset;
+import org.apache.inlong.sort.cdc.base.source.meta.offset.OffsetFactory;
+import org.apache.inlong.sort.cdc.base.source.meta.split.SourceSplitState;
+import org.apache.inlong.sort.cdc.base.source.metrics.SourceReaderMetrics;
+import org.apache.inlong.sort.cdc.base.source.reader.IncrementalSourceReader;
+import org.apache.inlong.sort.cdc.base.source.reader.IncrementalSourceRecordEmitter;
+import org.apache.inlong.sort.cdc.base.util.RecordUtils;
+
+import io.debezium.connector.AbstractSourceInfo;
+import io.debezium.data.Envelope;
+import io.debezium.document.Array;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.HistoryRecord;
+import io.debezium.relational.history.TableChanges;
+import io.debezium.relational.history.TableChanges.TableChange;
+import org.apache.flink.api.connector.source.SourceOutput;
+import org.apache.flink.connector.base.source.reader.RecordEmitter;
+import org.apache.flink.util.Collector;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+
+import static com.ververica.cdc.connectors.base.source.meta.wartermark.WatermarkEvent.isHighWatermarkEvent;
+import static com.ververica.cdc.connectors.base.source.meta.wartermark.WatermarkEvent.isWatermarkEvent;
+import static com.ververica.cdc.connectors.base.utils.SourceRecordUtils.getHistoryRecord;
+import static com.ververica.cdc.connectors.base.utils.SourceRecordUtils.isDataChangeRecord;
+import static com.ververica.cdc.connectors.base.utils.SourceRecordUtils.isSchemaChangeEvent;
Review Comment:
Fix dependency to `org.apache.inlong.sort.cdc`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] EMsnap commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "EMsnap (via GitHub)" <gi...@apache.org>.
EMsnap commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1294122567
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/pom.xml:
##########
@@ -29,14 +29,26 @@
<name>Apache InLong - Sort-connector-postgres-cdc</name>
<properties>
+ <postgress.debezium.version>1.6.4.Final</postgress.debezium.version>
<inlong.root.dir>${project.parent.parent.parent.parent.parent.basedir}</inlong.root.dir>
</properties>
<dependencies>
<dependency>
<groupId>com.ververica</groupId>
- <artifactId>flink-connector-postgres-cdc</artifactId>
+ <artifactId>flink-connector-debezium</artifactId>
+ <version>2.3.0</version>
Review Comment:
please add version here
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/pom.xml:
##########
@@ -56,32 +68,59 @@
<artifactId>audit-sdk</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.checkerframework</groupId>
+ <artifactId>checker-qual</artifactId>
+ <version>3.5.0</version>
+ </dependency>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-postgres</artifactId>
- <version>${debezium.version}</version>
+ <version>${postgress.debezium.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.postgresql</groupId>
+ <artifactId>postgresql</artifactId>
+ <version>42.4.1</version>
Review Comment:
ditto
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/source/options/SourceOptions.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.inlong.sort.cdc.postgres.source.options;
+
+import org.apache.inlong.sort.cdc.base.source.IncrementalSource;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/** Configurations for {@link IncrementalSource}. */
+public class SourceOptions {
Review Comment:
source options is not used here
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/pom.xml:
##########
@@ -29,14 +29,26 @@
<name>Apache InLong - Sort-connector-postgres-cdc</name>
<properties>
+ <postgress.debezium.version>1.6.4.Final</postgress.debezium.version>
<inlong.root.dir>${project.parent.parent.parent.parent.parent.basedir}</inlong.root.dir>
</properties>
<dependencies>
<dependency>
<groupId>com.ververica</groupId>
- <artifactId>flink-connector-postgres-cdc</artifactId>
+ <artifactId>flink-connector-debezium</artifactId>
+ <version>2.3.0</version>
</dependency>
+ <dependency>
+ <groupId>com.ververica</groupId>
+ <artifactId>flink-cdc-base</artifactId>
+ <version>2.3.0</version>
+ </dependency>
+ <!-- <dependency>-->
Review Comment:
Useless comment here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] e-mhui commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "e-mhui (via GitHub)" <gi...@apache.org>.
e-mhui commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1320937674
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/cdc/postgres/source/config/PostgresSourceConfig.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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.inlong.sort.cdc.postgres.source.config;
+
+import org.apache.inlong.sort.cdc.base.config.JdbcSourceConfig;
+
+import com.ververica.cdc.connectors.base.options.StartupOptions;
+import io.debezium.config.Configuration;
+import io.debezium.connector.AbstractSourceInfo;
+import io.debezium.connector.postgresql.PostgresConnectorConfig;
+import io.debezium.relational.RelationalTableFilters;
+
+import javax.annotation.Nullable;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+
+/** The configuration for Postgres CDC source. */
+public class PostgresSourceConfig extends JdbcSourceConfig {
+
+ private static final long serialVersionUID = 1L;
+
+ public PostgresSourceConfig(
+ StartupOptions startupOptions,
+ List<String> databaseList,
+ List<String> schemaList,
+ List<String> tableList,
+ int splitSize,
+ int splitMetaGroupSize,
+ double distributionFactorUpper,
+ double distributionFactorLower,
+ boolean includeSchemaChanges,
+ Properties dbzProperties,
+ Configuration dbzConfiguration,
+ String driverClassName,
+ String hostname,
+ int port,
+ String username,
+ String password,
+ int fetchSize,
+ String serverTimeZone,
+ Duration connectTimeout,
+ int connectMaxRetries,
+ int connectionPoolSize,
+ @Nullable String chunkKeyColumn,
+ String inlongMetric,
+ String inlongAudit) {
+ super(
+ startupOptions,
+ databaseList,
+ schemaList,
+ tableList,
+ splitSize,
+ splitMetaGroupSize,
+ distributionFactorUpper,
+ distributionFactorLower,
+ includeSchemaChanges,
+ dbzProperties,
+ dbzConfiguration,
+ driverClassName,
+ hostname,
+ port,
+ username,
+ password,
+ fetchSize,
+ serverTimeZone,
+ connectTimeout,
+ connectMaxRetries,
+ connectionPoolSize,
+ chunkKeyColumn,
+ inlongMetric,
+ inlongAudit);
+ }
+
+ @Override
+ public PostgresConnectorConfig getDbzConnectorConfig() {
+ return new PostgresConnectorConfig(getDbzConfiguration());
+ }
+
+ public RelationalTableFilters getTableFilters() {
+ return getDbzConnectorConfig().getTableFilters();
+ }
+
+ @Override
+ public List<String> getMetricLabelList() {
+ return Arrays.asList(AbstractSourceInfo.DATABASE_NAME_KEY,
Review Comment:
The key should be 'database', not 'db'.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] yunqingmoswu commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "yunqingmoswu (via GitHub)" <gi...@apache.org>.
yunqingmoswu commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1324398449
##########
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/HybridSplitAssigner.java:
##########
@@ -123,7 +123,8 @@ public Optional<SourceSplitBase> getNext() {
// assigning the stream split. Otherwise, records emitted from stream split
// might be out-of-order in terms of same primary key with snapshot splits.
isStreamSplitAssigned = true;
- return Optional.of(createStreamSplit());
+ StreamSplit streamSplit = createStreamSplit();
Review Comment:
This change is not necessary?
##########
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/oracle-cdc/src/main/java/org/apache/inlong/sort/cdc/oracle/source/config/OracleSourceConfigFactory.java:
##########
@@ -108,10 +108,6 @@ public OracleSourceConfig create(int subtaskId) {
props.setProperty("database.port", String.valueOf(port));
}
- if (schemaList != null) {
Review Comment:
Why change the code of Oracle cdc?
##########
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/reader/IncrementalSourceReader.java:
##########
@@ -237,7 +242,7 @@ private StreamSplit discoverTableSchemasForStreamSplit(StreamSplit split) {
public void handleSourceEvents(SourceEvent sourceEvent) {
if (sourceEvent instanceof FinishedSnapshotSplitsAckEvent) {
FinishedSnapshotSplitsAckEvent ackEvent = (FinishedSnapshotSplitsAckEvent) sourceEvent;
- LOG.debug(
+ LOG.info(
Review Comment:
Why change the log level, maybe it will cause a large amount of log output?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] kuansix commented on pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "kuansix (via GitHub)" <gi...@apache.org>.
kuansix commented on PR #8664:
URL: https://github.com/apache/inlong/pull/8664#issuecomment-1676305645
> @kuansix, please add more description for this PR.
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] EMsnap commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "EMsnap (via GitHub)" <gi...@apache.org>.
EMsnap commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1292942681
##########
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/source/assigner/state/PendingSplitsStateSerializer.java:
##########
@@ -369,8 +372,11 @@ private List<TableId> readTableIds(DataInputDeserializer in) throws IOException
List<TableId> tableIds = new ArrayList<>();
final int size = in.readInt();
for (int i = 0; i < size; i++) {
+ // boolean useCatalogBeforeSchema = in.readBoolean();
Review Comment:
remove useless comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] gong commented on a diff in pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "gong (via GitHub)" <gi...@apache.org>.
gong commented on code in PR #8664:
URL: https://github.com/apache/inlong/pull/8664#discussion_r1307201077
##########
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/relational/connection/JdbcConnectionPoolFactory.java:
##########
@@ -47,6 +47,8 @@ public HikariDataSource createPooledDataSource(JdbcSourceConfig sourceConfig) {
config.addDataSourceProperty(SERVER_TIMEZONE_KEY, sourceConfig.getServerTimeZone());
config.setDriverClassName(sourceConfig.getDriverClassName());
+ // note: the following properties should be optional (only applied to MySQL)
+ config.addDataSourceProperty(SERVER_TIMEZONE_KEY, sourceConfig.getServerTimeZone());
Review Comment:
`config.addDataSourceProperty(SERVER_TIMEZONE_KEY, sourceConfig.getServerTimeZone());` is duplication code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] kuansix commented on pull request #8664: [INLONG-7908][Sort] PostgreSQL connector supports parallel read
Posted by "kuansix (via GitHub)" <gi...@apache.org>.
kuansix commented on PR #8664:
URL: https://github.com/apache/inlong/pull/8664#issuecomment-1716906877
> Please add a test on parallel read and I guess some of the classes can reuse flink-cdc rather than copying them
1.add a test (done)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org