Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on PR #6303: URL: https://github.com/apache/gravitino/pull/6303#issuecomment-2611293599 @sunxiaojian thanks for your work to make code cleaner. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 merged PR #6303: URL: https://github.com/apache/gravitino/pull/6303 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on PR #6303: URL: https://github.com/apache/gravitino/pull/6303#issuecomment-2610162403 LGTM -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1927136709
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -65,7 +71,7 @@ void testCatalogPropertiesWithRestBackend() {
"rest-uri",
IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
"rest-warehouse",
-"key1",
Review Comment:
I have already refined and improved the 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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1927129847
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/hive/HivePropertiesConverter.java:
##
@@ -34,46 +31,19 @@ public class HivePropertiesConverter implements
PropertiesConverter {
private HivePropertiesConverter() {}
public static final HivePropertiesConverter INSTANCE = new
HivePropertiesConverter();
-
private static final Map HIVE_CATALOG_CONFIG_TO_GRAVITINO =
ImmutableMap.of(HiveConf.ConfVars.METASTOREURIS.varname,
HiveConstants.METASTORE_URIS);
private static final Map GRAVITINO_CONFIG_TO_HIVE =
ImmutableMap.of(HiveConstants.METASTORE_URIS,
HiveConf.ConfVars.METASTOREURIS.varname);
@Override
- public Map toGravitinoCatalogProperties(Configuration
flinkConf) {
-Map gravitinoProperties = Maps.newHashMap();
-
-for (Map.Entry entry : flinkConf.toMap().entrySet()) {
- String gravitinoKey =
HIVE_CATALOG_CONFIG_TO_GRAVITINO.get(entry.getKey());
- if (gravitinoKey != null) {
-gravitinoProperties.put(gravitinoKey, entry.getValue());
- } else if (!entry.getKey().startsWith(FLINK_PROPERTY_PREFIX)) {
-gravitinoProperties.put(FLINK_PROPERTY_PREFIX + entry.getKey(),
entry.getValue());
- } else {
-gravitinoProperties.put(entry.getKey(), entry.getValue());
- }
-}
-
-return gravitinoProperties;
+ public String transformPropertyToGravitinoCatalog(String configKey) {
+return HIVE_CATALOG_CONFIG_TO_GRAVITINO.get(configKey);
}
@Override
- public Map toFlinkCatalogProperties(Map
gravitinoProperties) {
-Map flinkCatalogProperties = Maps.newHashMap();
-flinkCatalogProperties.put(
-CommonCatalogOptions.CATALOG_TYPE.key(),
GravitinoHiveCatalogFactoryOptions.IDENTIFIER);
-
-gravitinoProperties.forEach(
-(key, value) -> {
- String flinkConfigKey = key;
- if (key.startsWith(PropertiesConverter.FLINK_PROPERTY_PREFIX)) {
-flinkConfigKey =
key.substring(PropertiesConverter.FLINK_PROPERTY_PREFIX.length());
- }
- flinkCatalogProperties.put(
- GRAVITINO_CONFIG_TO_HIVE.getOrDefault(flinkConfigKey,
flinkConfigKey), value);
-});
-return flinkCatalogProperties;
+ public String transformPropertyToFlinkCatalog(String configKey) {
+return GRAVITINO_CONFIG_TO_HIVE.getOrDefault(configKey, configKey);
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1927128065
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConverter.java:
##
@@ -38,38 +36,21 @@ private IcebergPropertiesConverter() {}
IcebergConstants.CATALOG_BACKEND,
IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE);
@Override
- public Map toFlinkCatalogProperties(Map
gravitinoProperties) {
-Preconditions.checkArgument(
-gravitinoProperties != null, "Iceberg Catalog properties should not be
null.");
+ public String transformPropertyToGravitinoCatalog(String configKey) {
+return
IcebergPropertiesUtils.ICEBERG_CATALOG_CONFIG_TO_GRAVITINO.get(configKey);
+ }
-Map all = new HashMap<>();
-if (gravitinoProperties != null) {
- gravitinoProperties.forEach(
- (k, v) -> {
-if (k.startsWith(FLINK_PROPERTY_PREFIX)) {
- String newKey = k.substring(FLINK_PROPERTY_PREFIX.length());
- all.put(newKey, v);
-}
- });
-}
-Map transformedProperties =
-IcebergPropertiesUtils.toIcebergCatalogProperties(gravitinoProperties);
+ @Override
+ public String transformPropertyToFlinkCatalog(String configKey) {
-if (transformedProperties != null) {
- all.putAll(transformedProperties);
+String icebergConfigKey = configKey;
+if
(IcebergPropertiesUtils.GRAVITINO_CONFIG_TO_ICEBERG.containsKey(configKey)) {
+ icebergConfigKey =
IcebergPropertiesUtils.GRAVITINO_CONFIG_TO_ICEBERG.get(configKey);
}
-all.put(
-CommonCatalogOptions.CATALOG_TYPE.key(),
GravitinoIcebergCatalogFactoryOptions.IDENTIFIER);
-// Map "catalog-backend" to "catalog-type".
-// TODO If catalog backend is CUSTOM, we need special compatibility logic.
-GRAVITINO_CONFIG_TO_FLINK_ICEBERG.forEach(
-(key, value) -> {
- if (all.containsKey(key)) {
-String config = all.remove(key);
-all.put(value, config);
- }
-});
-return all;
+if (GRAVITINO_CONFIG_TO_FLINK_ICEBERG.containsKey(configKey)) {
+ icebergConfigKey = GRAVITINO_CONFIG_TO_FLINK_ICEBERG.get(configKey);
+}
+return icebergConfigKey;
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1927121625
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -50,7 +52,11 @@ void testCatalogPropertiesWithHiveBackend() {
IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
"hive-uri",
IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
-"hive-warehouse"),
+"hive-warehouse",
Review Comment:
@FANNG1 Ohh, this is the logic before this modification. It seems that no
modification is needed at all. I just need to add some comments to clarify the
misunderstanding; there is no disagreement 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1927121625
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -50,7 +52,11 @@ void testCatalogPropertiesWithHiveBackend() {
IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
"hive-uri",
IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
-"hive-warehouse"),
+"hive-warehouse",
Review Comment:
@FANNG1 Ohh, this is the logic before this modification.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1927033463
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -50,7 +52,11 @@ void testCatalogPropertiesWithHiveBackend() {
IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
"hive-uri",
IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
-"hive-warehouse"),
+"hive-warehouse",
Review Comment:
revert
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1927112224
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/hive/HivePropertiesConverter.java:
##
@@ -34,46 +31,19 @@ public class HivePropertiesConverter implements
PropertiesConverter {
private HivePropertiesConverter() {}
public static final HivePropertiesConverter INSTANCE = new
HivePropertiesConverter();
-
private static final Map HIVE_CATALOG_CONFIG_TO_GRAVITINO =
ImmutableMap.of(HiveConf.ConfVars.METASTOREURIS.varname,
HiveConstants.METASTORE_URIS);
private static final Map GRAVITINO_CONFIG_TO_HIVE =
ImmutableMap.of(HiveConstants.METASTORE_URIS,
HiveConf.ConfVars.METASTOREURIS.varname);
@Override
- public Map toGravitinoCatalogProperties(Configuration
flinkConf) {
-Map gravitinoProperties = Maps.newHashMap();
-
-for (Map.Entry entry : flinkConf.toMap().entrySet()) {
- String gravitinoKey =
HIVE_CATALOG_CONFIG_TO_GRAVITINO.get(entry.getKey());
- if (gravitinoKey != null) {
-gravitinoProperties.put(gravitinoKey, entry.getValue());
- } else if (!entry.getKey().startsWith(FLINK_PROPERTY_PREFIX)) {
-gravitinoProperties.put(FLINK_PROPERTY_PREFIX + entry.getKey(),
entry.getValue());
- } else {
-gravitinoProperties.put(entry.getKey(), entry.getValue());
- }
-}
-
-return gravitinoProperties;
+ public String transformPropertyToGravitinoCatalog(String configKey) {
+return HIVE_CATALOG_CONFIG_TO_GRAVITINO.get(configKey);
}
@Override
- public Map toFlinkCatalogProperties(Map
gravitinoProperties) {
-Map flinkCatalogProperties = Maps.newHashMap();
-flinkCatalogProperties.put(
-CommonCatalogOptions.CATALOG_TYPE.key(),
GravitinoHiveCatalogFactoryOptions.IDENTIFIER);
-
-gravitinoProperties.forEach(
-(key, value) -> {
- String flinkConfigKey = key;
- if (key.startsWith(PropertiesConverter.FLINK_PROPERTY_PREFIX)) {
-flinkConfigKey =
key.substring(PropertiesConverter.FLINK_PROPERTY_PREFIX.length());
- }
- flinkCatalogProperties.put(
- GRAVITINO_CONFIG_TO_HIVE.getOrDefault(flinkConfigKey,
flinkConfigKey), value);
-});
-return flinkCatalogProperties;
+ public String transformPropertyToFlinkCatalog(String configKey) {
+return GRAVITINO_CONFIG_TO_HIVE.getOrDefault(configKey, configKey);
Review Comment:
`transformPropertyToFlinkCatalog` return null if the config key not in the
map, WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian closed pull request #6303: [#6302] refactor:Optimize Flink connector properties converter URL: https://github.com/apache/gravitino/pull/6303 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1927033463
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -50,7 +52,11 @@ void testCatalogPropertiesWithHiveBackend() {
IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
"hive-uri",
IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
-"hive-warehouse"),
+"hive-warehouse",
Review Comment:
revert
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1927030277
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/hive/TestHivePropertiesConverter.java:
##
@@ -35,7 +35,7 @@ public void testToGravitinoCatalogProperties() {
Configuration configuration =
Configuration.fromMap(
ImmutableMap.of(
-"hive-conf-dir",
Review Comment:
revert
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1927030921
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java:
##
@@ -32,25 +34,80 @@ public interface PropertiesConverter {
/**
* Converts properties from application provided properties and Flink
connector properties to
- * Gravitino properties.
+ * Gravitino properties.This method processes the Flink configuration and
transforms it into a
+ * format suitable for the Gravitino catalog.
*
- * @param flinkConf The configuration provided by Flink.
- * @return properties for the Gravitino catalog.
+ * @param flinkConf The Flink configuration containing connector properties.
This includes both
+ * Flink-specific properties and any user-provided properties.
+ * @return A map of properties converted for use in the Gravitino catalog.
The returned map
+ * includes both directly transformed properties and bypass properties
prefixed with {@link
+ * #FLINK_PROPERTY_PREFIX}.
*/
default Map toGravitinoCatalogProperties(Configuration
flinkConf) {
-return flinkConf.toMap();
+Map gravitinoProperties = Maps.newHashMap();
+for (Map.Entry entry : flinkConf.toMap().entrySet()) {
+ String gravitinoKey =
transformPropertyToGravitinoCatalog(entry.getKey());
+ if (gravitinoKey != null) {
+gravitinoProperties.put(gravitinoKey, entry.getValue());
+ } else if (entry.getKey().startsWith(FLINK_PROPERTY_PREFIX)) {
+gravitinoProperties.put(entry.getKey(), entry.getValue());
+ }
Review Comment:
revert
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1927026802
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java:
##
@@ -32,25 +34,80 @@ public interface PropertiesConverter {
/**
* Converts properties from application provided properties and Flink
connector properties to
- * Gravitino properties.
+ * Gravitino properties.This method processes the Flink configuration and
transforms it into a
+ * format suitable for the Gravitino catalog.
*
- * @param flinkConf The configuration provided by Flink.
- * @return properties for the Gravitino catalog.
+ * @param flinkConf The Flink configuration containing connector properties.
This includes both
+ * Flink-specific properties and any user-provided properties.
+ * @return A map of properties converted for use in the Gravitino catalog.
The returned map
+ * includes both directly transformed properties and bypass properties
prefixed with {@link
+ * #FLINK_PROPERTY_PREFIX}.
*/
default Map toGravitinoCatalogProperties(Configuration
flinkConf) {
-return flinkConf.toMap();
+Map gravitinoProperties = Maps.newHashMap();
+for (Map.Entry entry : flinkConf.toMap().entrySet()) {
+ String gravitinoKey =
transformPropertyToGravitinoCatalog(entry.getKey());
+ if (gravitinoKey != null) {
+gravitinoProperties.put(gravitinoKey, entry.getValue());
+ } else if (entry.getKey().startsWith(FLINK_PROPERTY_PREFIX)) {
+gravitinoProperties.put(entry.getKey(), entry.getValue());
+ }
Review Comment:
If that's the case, I don't need to make this modification
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1926286527
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -50,7 +52,11 @@ void testCatalogPropertiesWithHiveBackend() {
IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
"hive-uri",
IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
-"hive-warehouse"),
+"hive-warehouse",
Review Comment:
@sunxiaojian , how about following the logics here?
When creating a catalog, the properties managed by Gravitino are
transformed, the properties starting with `flink.bypass` are left, and the
other properties will add the `flink.bypass` prefix.
When loading a catalog, the properties managed by Gravitino are
transformed, the properties starting with `flink.bypass` are removed from the
prefix, and others are dropped.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1926282137
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/hive/TestHivePropertiesConverter.java:
##
@@ -35,7 +35,7 @@ public void testToGravitinoCatalogProperties() {
Configuration configuration =
Configuration.fromMap(
ImmutableMap.of(
-"hive-conf-dir",
Review Comment:
better to pass `hive-conf-dir` not `flink.bypass.hive-conf-dir` when
creating a hive catalog from Flink SQL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1926280075
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/IcebergPropertiesConverter.java:
##
@@ -38,38 +36,21 @@ private IcebergPropertiesConverter() {}
IcebergConstants.CATALOG_BACKEND,
IcebergPropertiesConstants.ICEBERG_CATALOG_TYPE);
@Override
- public Map toFlinkCatalogProperties(Map
gravitinoProperties) {
-Preconditions.checkArgument(
-gravitinoProperties != null, "Iceberg Catalog properties should not be
null.");
+ public String transformPropertyToGravitinoCatalog(String configKey) {
+return
IcebergPropertiesUtils.ICEBERG_CATALOG_CONFIG_TO_GRAVITINO.get(configKey);
+ }
-Map all = new HashMap<>();
-if (gravitinoProperties != null) {
- gravitinoProperties.forEach(
- (k, v) -> {
-if (k.startsWith(FLINK_PROPERTY_PREFIX)) {
- String newKey = k.substring(FLINK_PROPERTY_PREFIX.length());
- all.put(newKey, v);
-}
- });
-}
-Map transformedProperties =
-IcebergPropertiesUtils.toIcebergCatalogProperties(gravitinoProperties);
+ @Override
+ public String transformPropertyToFlinkCatalog(String configKey) {
-if (transformedProperties != null) {
- all.putAll(transformedProperties);
+String icebergConfigKey = configKey;
+if
(IcebergPropertiesUtils.GRAVITINO_CONFIG_TO_ICEBERG.containsKey(configKey)) {
+ icebergConfigKey =
IcebergPropertiesUtils.GRAVITINO_CONFIG_TO_ICEBERG.get(configKey);
}
-all.put(
-CommonCatalogOptions.CATALOG_TYPE.key(),
GravitinoIcebergCatalogFactoryOptions.IDENTIFIER);
-// Map "catalog-backend" to "catalog-type".
-// TODO If catalog backend is CUSTOM, we need special compatibility logic.
-GRAVITINO_CONFIG_TO_FLINK_ICEBERG.forEach(
-(key, value) -> {
- if (all.containsKey(key)) {
-String config = all.remove(key);
-all.put(value, config);
- }
-});
-return all;
+if (GRAVITINO_CONFIG_TO_FLINK_ICEBERG.containsKey(configKey)) {
+ icebergConfigKey = GRAVITINO_CONFIG_TO_FLINK_ICEBERG.get(configKey);
+}
+return icebergConfigKey;
Review Comment:
how about return null if neither of the maps contains the configKey?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1926279677
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/hive/HivePropertiesConverter.java:
##
@@ -34,46 +31,19 @@ public class HivePropertiesConverter implements
PropertiesConverter {
private HivePropertiesConverter() {}
public static final HivePropertiesConverter INSTANCE = new
HivePropertiesConverter();
-
private static final Map HIVE_CATALOG_CONFIG_TO_GRAVITINO =
ImmutableMap.of(HiveConf.ConfVars.METASTOREURIS.varname,
HiveConstants.METASTORE_URIS);
private static final Map GRAVITINO_CONFIG_TO_HIVE =
ImmutableMap.of(HiveConstants.METASTORE_URIS,
HiveConf.ConfVars.METASTOREURIS.varname);
@Override
- public Map toGravitinoCatalogProperties(Configuration
flinkConf) {
-Map gravitinoProperties = Maps.newHashMap();
-
-for (Map.Entry entry : flinkConf.toMap().entrySet()) {
- String gravitinoKey =
HIVE_CATALOG_CONFIG_TO_GRAVITINO.get(entry.getKey());
- if (gravitinoKey != null) {
-gravitinoProperties.put(gravitinoKey, entry.getValue());
- } else if (!entry.getKey().startsWith(FLINK_PROPERTY_PREFIX)) {
-gravitinoProperties.put(FLINK_PROPERTY_PREFIX + entry.getKey(),
entry.getValue());
- } else {
-gravitinoProperties.put(entry.getKey(), entry.getValue());
- }
-}
-
-return gravitinoProperties;
+ public String transformPropertyToGravitinoCatalog(String configKey) {
+return HIVE_CATALOG_CONFIG_TO_GRAVITINO.get(configKey);
}
@Override
- public Map toFlinkCatalogProperties(Map
gravitinoProperties) {
-Map flinkCatalogProperties = Maps.newHashMap();
-flinkCatalogProperties.put(
-CommonCatalogOptions.CATALOG_TYPE.key(),
GravitinoHiveCatalogFactoryOptions.IDENTIFIER);
-
-gravitinoProperties.forEach(
-(key, value) -> {
- String flinkConfigKey = key;
- if (key.startsWith(PropertiesConverter.FLINK_PROPERTY_PREFIX)) {
-flinkConfigKey =
key.substring(PropertiesConverter.FLINK_PROPERTY_PREFIX.length());
- }
- flinkCatalogProperties.put(
- GRAVITINO_CONFIG_TO_HIVE.getOrDefault(flinkConfigKey,
flinkConfigKey), value);
-});
-return flinkCatalogProperties;
+ public String transformPropertyToFlinkCatalog(String configKey) {
+return GRAVITINO_CONFIG_TO_HIVE.getOrDefault(configKey, configKey);
Review Comment:
How about using `GRAVITINO_CONFIG_TO_HIVE.get(configKey)`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1926278338
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java:
##
@@ -32,25 +34,80 @@ public interface PropertiesConverter {
/**
* Converts properties from application provided properties and Flink
connector properties to
- * Gravitino properties.
+ * Gravitino properties.This method processes the Flink configuration and
transforms it into a
+ * format suitable for the Gravitino catalog.
*
- * @param flinkConf The configuration provided by Flink.
- * @return properties for the Gravitino catalog.
+ * @param flinkConf The Flink configuration containing connector properties.
This includes both
+ * Flink-specific properties and any user-provided properties.
+ * @return A map of properties converted for use in the Gravitino catalog.
The returned map
+ * includes both directly transformed properties and bypass properties
prefixed with {@link
+ * #FLINK_PROPERTY_PREFIX}.
*/
default Map toGravitinoCatalogProperties(Configuration
flinkConf) {
-return flinkConf.toMap();
+Map gravitinoProperties = Maps.newHashMap();
+for (Map.Entry entry : flinkConf.toMap().entrySet()) {
+ String gravitinoKey =
transformPropertyToGravitinoCatalog(entry.getKey());
+ if (gravitinoKey != null) {
+gravitinoProperties.put(gravitinoKey, entry.getValue());
+ } else if (entry.getKey().startsWith(FLINK_PROPERTY_PREFIX)) {
+gravitinoProperties.put(entry.getKey(), entry.getValue());
+ }
Review Comment:
if user create a catalog with properties not defined by gravitino or start
with `flink.bypass.`, we should add `flink.bypass.` to it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1926220148
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -50,7 +52,11 @@ void testCatalogPropertiesWithHiveBackend() {
IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
"hive-uri",
IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
-"hive-warehouse"),
+"hive-warehouse",
Review Comment:
Got it, the origin properties transform logic in Hive maybe bugy, cc
@coolderli
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1925562999
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -50,7 +52,11 @@ void testCatalogPropertiesWithHiveBackend() {
IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
"hive-uri",
IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
-"hive-warehouse"),
+"hive-warehouse",
Review Comment:
The current unified logic is to automatically add a prefix to parameters
that have not been mapped and have not been added with "flink.bypass." before
passing them to Gravitino. When setting them, there is no mandatory restriction
that they must include "flink.bypass, This is also for compatibility with old
logic, but what you said is also reasonable. Now we need to make some
adjustments to the integration testing of Hive
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1925562999
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -50,7 +52,11 @@ void testCatalogPropertiesWithHiveBackend() {
IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
"hive-uri",
IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
-"hive-warehouse"),
+"hive-warehouse",
Review Comment:
The current unified logic is to automatically add a prefix to parameters
that have not been mapped and have not been added with "Flink. bypass." before
passing them to Gravitino. When setting them, there is no mandatory restriction
that they must include "Flink. bypass, This is also for compatibility with old
logic, but what you said is also reasonable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1925562999
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -50,7 +52,11 @@ void testCatalogPropertiesWithHiveBackend() {
IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
"hive-uri",
IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
-"hive-warehouse"),
+"hive-warehouse",
Review Comment:
The current unified logic is to automatically add a prefix to parameters
that have not been mapped and have not been added with "flink.bypass." before
passing them to Gravitino. When setting them, there is no mandatory restriction
that they must include "flink.bypass, This is also for compatibility with old
logic, but what you said is also reasonable. Let me unify it first
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1925562999
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -50,7 +52,11 @@ void testCatalogPropertiesWithHiveBackend() {
IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
"hive-uri",
IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
-"hive-warehouse"),
+"hive-warehouse",
Review Comment:
The current unified logic is to automatically add a prefix to parameters
that have not been mapped and have not been added with "Flink. bypass." before
passing them to Gravitino. When setting them, there is no mandatory restriction
that they must include "Flink. bypass, This is also for compatibility with old
logic, but what you said is also reasonable. Let me unify it first
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on PR #6303: URL: https://github.com/apache/gravitino/pull/6303#issuecomment-2606224158 LGTM except minor comments -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1924642525
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -65,7 +71,7 @@ void testCatalogPropertiesWithRestBackend() {
"rest-uri",
IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
"rest-warehouse",
-"key1",
Review Comment:
could you use `key1` and `flink.bypass.key2` too?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1924642142
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -50,7 +52,11 @@ void testCatalogPropertiesWithHiveBackend() {
IcebergPropertiesConstants.ICEBERG_CATALOG_URI,
"hive-uri",
IcebergPropertiesConstants.ICEBERG_CATALOG_WAREHOUSE,
-"hive-warehouse"),
+"hive-warehouse",
Review Comment:
key1 and value1 should not exists in the result. By design only the keys
defined by Graviitno like `catalog-backend` or start with `flink.bypass.` are
passed to the underlying flink connector. the other property keys are dropped
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1924601566
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -39,7 +39,7 @@ void testCatalogPropertiesWithHiveBackend() {
"hive-uri",
IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
"hive-warehouse",
-"key1",
Review Comment:
fixed
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java:
##
@@ -32,25 +34,81 @@ public interface PropertiesConverter {
/**
* Converts properties from application provided properties and Flink
connector properties to
- * Gravitino properties.
+ * Gravitino properties.This method processes the Flink configuration and
transforms it into a
+ * format suitable for the Gravitino catalog.
*
- * @param flinkConf The configuration provided by Flink.
- * @return properties for the Gravitino catalog.
+ * @param flinkConf The Flink configuration containing connector properties.
This includes both
+ * Flink-specific properties and any user-provided properties.
+ * @return A map of properties converted for use in the Gravitino catalog.
The returned map
+ * includes both directly transformed properties and bypass properties
prefixed with {@link
+ * #FLINK_PROPERTY_PREFIX}.
*/
default Map toGravitinoCatalogProperties(Configuration
flinkConf) {
-return flinkConf.toMap();
+Map gravitinoProperties = Maps.newHashMap();
+for (Map.Entry entry : flinkConf.toMap().entrySet()) {
+ String gravitinoKey =
transformPropertyToGravitinoCatalog(entry.getKey());
+ if (gravitinoKey != null) {
+gravitinoProperties.put(gravitinoKey, entry.getValue());
+ } else if (!entry.getKey().startsWith(FLINK_PROPERTY_PREFIX)) {
+gravitinoProperties.put(FLINK_PROPERTY_PREFIX + entry.getKey(),
entry.getValue());
+ } else {
+gravitinoProperties.put(entry.getKey(), entry.getValue());
+ }
+}
+return gravitinoProperties;
}
/**
- * Converts properties from Gravitino properties to Flink connector
properties.
+ * Converts properties from Gravitino catalog properties to Flink connector
properties. This
+ * method processes the Gravitino properties and transforms them into a
format suitable for the
+ * Flink connector.
*
- * @param gravitinoProperties The properties provided by Gravitino.
- * @return properties for the Flink connector.
+ * @param gravitinoProperties The properties provided by the Gravitino
catalog. This includes both
+ * Gravitino-specific properties and any bypass properties prefixed with
{@link
+ * #FLINK_PROPERTY_PREFIX}.
+ * @return A map of properties converted for use in the Flink connector. The
returned map includes
+ * both transformed properties and the Flink catalog type.
*/
default Map toFlinkCatalogProperties(Map
gravitinoProperties) {
-return gravitinoProperties;
+Map allProperties = Maps.newHashMap();
+gravitinoProperties.forEach(
+(key, value) -> {
+ String flinkConfigKey = key;
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1923795155
##
flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/iceberg/TestIcebergPropertiesConverter.java:
##
@@ -39,7 +39,7 @@ void testCatalogPropertiesWithHiveBackend() {
"hive-uri",
IcebergPropertiesConstants.GRAVITINO_ICEBERG_CATALOG_WAREHOUSE,
"hive-warehouse",
-"key1",
Review Comment:
`key1` and `value1` is used to check the key not in converted key map will
not in the transformed properties. You could add `flink.bypass.key2` and
`value2`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1923792459
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java:
##
@@ -32,25 +34,81 @@ public interface PropertiesConverter {
/**
* Converts properties from application provided properties and Flink
connector properties to
- * Gravitino properties.
+ * Gravitino properties.This method processes the Flink configuration and
transforms it into a
+ * format suitable for the Gravitino catalog.
*
- * @param flinkConf The configuration provided by Flink.
- * @return properties for the Gravitino catalog.
+ * @param flinkConf The Flink configuration containing connector properties.
This includes both
+ * Flink-specific properties and any user-provided properties.
+ * @return A map of properties converted for use in the Gravitino catalog.
The returned map
+ * includes both directly transformed properties and bypass properties
prefixed with {@link
+ * #FLINK_PROPERTY_PREFIX}.
*/
default Map toGravitinoCatalogProperties(Configuration
flinkConf) {
-return flinkConf.toMap();
+Map gravitinoProperties = Maps.newHashMap();
+for (Map.Entry entry : flinkConf.toMap().entrySet()) {
+ String gravitinoKey =
transformPropertyToGravitinoCatalog(entry.getKey());
+ if (gravitinoKey != null) {
+gravitinoProperties.put(gravitinoKey, entry.getValue());
+ } else if (!entry.getKey().startsWith(FLINK_PROPERTY_PREFIX)) {
+gravitinoProperties.put(FLINK_PROPERTY_PREFIX + entry.getKey(),
entry.getValue());
+ } else {
+gravitinoProperties.put(entry.getKey(), entry.getValue());
+ }
+}
+return gravitinoProperties;
}
/**
- * Converts properties from Gravitino properties to Flink connector
properties.
+ * Converts properties from Gravitino catalog properties to Flink connector
properties. This
+ * method processes the Gravitino properties and transforms them into a
format suitable for the
+ * Flink connector.
*
- * @param gravitinoProperties The properties provided by Gravitino.
- * @return properties for the Flink connector.
+ * @param gravitinoProperties The properties provided by the Gravitino
catalog. This includes both
+ * Gravitino-specific properties and any bypass properties prefixed with
{@link
+ * #FLINK_PROPERTY_PREFIX}.
+ * @return A map of properties converted for use in the Flink connector. The
returned map includes
+ * both transformed properties and the Flink catalog type.
*/
default Map toFlinkCatalogProperties(Map
gravitinoProperties) {
-return gravitinoProperties;
+Map allProperties = Maps.newHashMap();
+gravitinoProperties.forEach(
+(key, value) -> {
+ String flinkConfigKey = key;
Review Comment:
seems this logic is correct, if the key start with flink.bypass, we should
extract the key, else we get the converted key from
`transformPropertyToFlinkCatalog`
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java:
##
@@ -32,25 +34,81 @@ public interface PropertiesConverter {
/**
* Converts properties from application provided properties and Flink
connector properties to
- * Gravitino properties.
+ * Gravitino properties.This method processes the Flink configuration and
transforms it into a
+ * format suitable for the Gravitino catalog.
*
- * @param flinkConf The configuration provided by Flink.
- * @return properties for the Gravitino catalog.
+ * @param flinkConf The Flink configuration containing connector properties.
This includes both
+ * Flink-specific properties and any user-provided properties.
+ * @return A map of properties converted for use in the Gravitino catalog.
The returned map
+ * includes both directly transformed properties and bypass properties
prefixed with {@link
+ * #FLINK_PROPERTY_PREFIX}.
*/
default Map toGravitinoCatalogProperties(Configuration
flinkConf) {
-return flinkConf.toMap();
+Map gravitinoProperties = Maps.newHashMap();
+for (Map.Entry entry : flinkConf.toMap().entrySet()) {
+ String gravitinoKey =
transformPropertyToGravitinoCatalog(entry.getKey());
+ if (gravitinoKey != null) {
+gravitinoProperties.put(gravitinoKey, entry.getValue());
+ } else if (!entry.getKey().startsWith(FLINK_PROPERTY_PREFIX)) {
+gravitinoProperties.put(FLINK_PROPERTY_PREFIX + entry.getKey(),
entry.getValue());
+ } else {
+gravitinoProperties.put(entry.getKey(), entry.getValue());
+ }
+}
+return gravitinoProperties;
}
/**
- * Converts properties from Gravitino properties to Flink connector
properties.
+ * Converts properties from Gravitino catalog properties to Flink connector
properties. This
+ * method processes the Gravitino proper
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1923147199
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java:
##
@@ -32,23 +34,76 @@ public interface PropertiesConverter {
/**
* Converts properties from application provided properties and Flink
connector properties to
- * Gravitino properties.
+ * Gravitino properties.This method processes the Flink configuration and
transforms it into a
+ * format suitable for the Gravitino catalog.
*
- * @param flinkConf The configuration provided by Flink.
- * @return properties for the Gravitino catalog.
+ * @param flinkConf The Flink configuration containing connector properties.
This includes both
+ * Flink-specific properties and any user-provided properties.
+ * @return A map of properties converted for use in the Gravitino catalog.
The returned map
+ * includes both directly transformed properties and bypass properties
prefixed with {@link
+ * #FLINK_PROPERTY_PREFIX}.
*/
default Map toGravitinoCatalogProperties(Configuration
flinkConf) {
-return flinkConf.toMap();
+Map gravitinoProperties = Maps.newHashMap();
+for (Map.Entry entry : flinkConf.toMap().entrySet()) {
+ String gravitinoKey =
transformPropertyToGravitinoCatalog(entry.getKey());
+ if (gravitinoKey != null) {
+gravitinoProperties.put(gravitinoKey, entry.getValue());
+ } else if (!entry.getKey().startsWith(FLINK_PROPERTY_PREFIX)) {
+gravitinoProperties.put(FLINK_PROPERTY_PREFIX + entry.getKey(),
entry.getValue());
+ } else {
+gravitinoProperties.put(entry.getKey(), entry.getValue());
+ }
+}
+return gravitinoProperties;
}
/**
- * Converts properties from Gravitino properties to Flink connector
properties.
+ * Converts properties from Gravitino catalog properties to Flink connector
properties. This
+ * method processes the Gravitino properties and transforms them into a
format suitable for the
+ * Flink connector.
*
- * @param gravitinoProperties The properties provided by Gravitino.
- * @return properties for the Flink connector.
+ * @param gravitinoProperties The properties provided by the Gravitino
catalog. This includes both
+ * Gravitino-specific properties and any bypass properties prefixed with
{@link
+ * #FLINK_PROPERTY_PREFIX}.
+ * @return A map of properties converted for use in the Flink connector. The
returned map includes
+ * both transformed properties and the Flink catalog type.
*/
default Map toFlinkCatalogProperties(Map
gravitinoProperties) {
-return gravitinoProperties;
+Map all = Maps.newHashMap();
+gravitinoProperties.forEach(
+(key, value) -> {
+ String flinkConfigKey = key;
+ if (key.startsWith(PropertiesConverter.FLINK_PROPERTY_PREFIX)) {
+flinkConfigKey =
key.substring(PropertiesConverter.FLINK_PROPERTY_PREFIX.length());
+ }
+ all.put(flinkConfigKey, value);
+});
+Map allProperties = transformPropertiesToFlinkCatalog(all);
+allProperties.put(CommonCatalogOptions.CATALOG_TYPE.key(),
getFlinkCatalogType());
+return allProperties;
+ }
+
+ /**
+ * Transforms a Flink configuration key to a corresponding Gravitino catalog
property key. This
+ * method is used to map Flink-specific configuration keys to Gravitino
catalog properties.
+ *
+ * @param configKey The Flink configuration key to be transformed.
+ * @return The corresponding Gravitino catalog property key, or {@code null}
if no transformation
+ * is needed.
+ */
+ String transformPropertyToGravitinoCatalog(String configKey);
+
+ /**
+ * Transforms a map of properties from Gravitino catalog properties to Flink
connector properties.
+ * This method is used to convert properties that are specific to Gravitino
into a format that can
+ * be understood by the Flink connector.
+ *
+ * @param allProperties A map of properties to be transformed.
+ * @return A map of properties transformed into Flink connector properties.
+ */
+ default Map transformPropertiesToFlinkCatalog(Map allProperties) {
Review Comment:
could we pass single gravitino property instead all properties as method
argument?
```
String transformPropertyToFlinkCatalog(String gravitinoProperty);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on PR #6303: URL: https://github.com/apache/gravitino/pull/6303#issuecomment-2603599733 @FANNG1 Fixed all ， PTAL again, 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1922631809
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/PaimonPropertiesConverter.java:
##
@@ -37,44 +34,38 @@ private PaimonPropertiesConverter() {}
@Override
public Map toGravitinoCatalogProperties(Configuration
flinkConf) {
Review Comment:
fixed
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java:
##
@@ -48,7 +61,24 @@ default Map
toGravitinoCatalogProperties(Configuration flinkConf
* @return properties for the Flink connector.
*/
default Map toFlinkCatalogProperties(Map
gravitinoProperties) {
-return gravitinoProperties;
+Map all = Maps.newHashMap();
+gravitinoProperties.forEach(
+(key, value) -> {
+ String flinkConfigKey = key;
+ if (key.startsWith(PropertiesConverter.FLINK_PROPERTY_PREFIX)) {
+flinkConfigKey =
key.substring(PropertiesConverter.FLINK_PROPERTY_PREFIX.length());
+ }
+ all.put(flinkConfigKey, value);
+});
+Map allProperties = transformPropertiesToFlinkCatalog(all);
+allProperties.put(CommonCatalogOptions.CATALOG_TYPE.key(),
getFlinkCatalogType());
+return allProperties;
+ }
+
+ String transformPropertiesToGravitinoCatalog(String configKey);
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1922149147
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java:
##
@@ -48,7 +61,24 @@ default Map
toGravitinoCatalogProperties(Configuration flinkConf
* @return properties for the Flink connector.
*/
default Map toFlinkCatalogProperties(Map
gravitinoProperties) {
-return gravitinoProperties;
+Map all = Maps.newHashMap();
+gravitinoProperties.forEach(
+(key, value) -> {
+ String flinkConfigKey = key;
+ if (key.startsWith(PropertiesConverter.FLINK_PROPERTY_PREFIX)) {
+flinkConfigKey =
key.substring(PropertiesConverter.FLINK_PROPERTY_PREFIX.length());
+ }
+ all.put(flinkConfigKey, value);
+});
+Map allProperties = transformPropertiesToFlinkCatalog(all);
+allProperties.put(CommonCatalogOptions.CATALOG_TYPE.key(),
getFlinkCatalogType());
+return allProperties;
+ }
+
+ String transformPropertiesToGravitinoCatalog(String configKey);
Review Comment:
Could you add Java doc for new interfaces? and please polish existing Java
doc for `toFlinkCatalogProperties` and `toGravitinoCatalogProperties`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1922146350
##
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/PaimonPropertiesConverter.java:
##
@@ -37,44 +34,38 @@ private PaimonPropertiesConverter() {}
@Override
public Map toGravitinoCatalogProperties(Configuration
flinkConf) {
Review Comment:
is `toGravitinoCatalogProperties` and `toFlinkCatalogProperties` still
usable in PaimonPropertiesConverter?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on PR #6303: URL: https://github.com/apache/gravitino/pull/6303#issuecomment-2597385437 @FANNG1 PTAL ， 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] [#6302] refactor:Optimize Flink connector properties converter [gravitino]
sunxiaojian commented on PR #6303: URL: https://github.com/apache/gravitino/pull/6303#issuecomment-2596188457 fixed : https://github.com/apache/gravitino/pull/6305 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
