[FLINK-24599][table] Replace static methods with convenient methods in LogicalType
Make the code of calling `hasRoot`/`hasFamily` less verbose by
replacing the static utility methods with member methods
`LogicalType#is()`.
This closes #17550.
diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaConnectorOptionsUtil.java b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaConnectorOptionsUtil.java
index 5514a5c..0c933ca 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaConnectorOptionsUtil.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaConnectorOptionsUtil.java
@@ -66,7 +66,6 @@
import static org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.VALUE_FIELDS_INCLUDE;
import static org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions.VALUE_FORMAT;
import static org.apache.flink.table.factories.FactoryUtil.FORMAT;
-import static org.apache.flink.table.types.logical.utils.LogicalTypeChecks.hasRoot;
/** Utilities for {@link KafkaConnectorOptions}. */
@Internal
@@ -412,7 +411,7 @@
ReadableConfig options, DataType physicalDataType) {
final LogicalType physicalType = physicalDataType.getLogicalType();
Preconditions.checkArgument(
- hasRoot(physicalType, LogicalTypeRoot.ROW), "Row data type expected.");
+ physicalType.is(LogicalTypeRoot.ROW), "Row data type expected.");
final Optional<String> optionalKeyFormat = options.getOptional(KEY_FORMAT);
final Optional<List<String>> optionalKeyFields = options.getOptional(KEY_FIELDS);
@@ -479,7 +478,7 @@
ReadableConfig options, DataType physicalDataType) {
final LogicalType physicalType = physicalDataType.getLogicalType();
Preconditions.checkArgument(
- hasRoot(physicalType, LogicalTypeRoot.ROW), "Row data type expected.");
+ physicalType.is(LogicalTypeRoot.ROW), "Row data type expected.");
final int physicalFieldCount = LogicalTypeChecks.getFieldCount(physicalType);
final IntStream physicalFields = IntStream.range(0, physicalFieldCount);
