From 322a841694968b5b8f7bb7c6cd9dd85d9d17e23f Mon Sep 17 00:00:00 2001
From: Jeyhun Karimov <je.karimov@gmail.com>
Date: Mon, 19 Feb 2024 15:16:02 +0100
Subject: [PATCH] [FLINK-34469][table] Implement TableDistribution toString

---
 .../java/org/apache/flink/table/api/TableDescriptor.java   | 3 +--
 .../apache/flink/table/api/internal/ShowCreateUtil.java    | 2 +-
 .../org/apache/flink/table/test/program/TableTestStep.java | 4 +---
 .../org/apache/flink/table/catalog/TableDistribution.java  | 7 ++++++-
 4 files changed, 9 insertions(+), 7 deletions(-)

diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableDescriptor.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableDescriptor.java
index 906cdafdf0d3b..42878bf1dcbe1 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableDescriptor.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableDescriptor.java
@@ -154,8 +154,7 @@ public String toString() {
                         .map(EncodingUtils::escapeIdentifier)
                         .collect(Collectors.joining(", "));
 
-        final String distributedBy =
-                distribution == null ? "" : distribution.asSerializableString();
+        final String distributedBy = distribution == null ? "" : distribution.toString();
 
         final String partitionedBy =
                 !partitionKeys.isEmpty()
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
index 36ad89176b26b..2a9eb90bbe75f 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/ShowCreateUtil.java
@@ -200,7 +200,7 @@ static Optional<String> extractFormattedComment(ResolvedCatalogBaseTable<?> tabl
     }
 
     static Optional<String> extractFormattedDistributedInfo(ResolvedCatalogTable catalogTable) {
-        return catalogTable.getDistribution().map(TableDistribution::asSerializableString);
+        return catalogTable.getDistribution().map(TableDistribution::toString);
     }
 
     static Optional<String> extractFormattedPartitionedInfo(ResolvedCatalogTable catalogTable) {
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestStep.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestStep.java
index 5d05bcec9f6e2..d63d11b08210a 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestStep.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/TableTestStep.java
@@ -66,9 +66,7 @@ public TableResult apply(TableEnvironment env, Map<String, String> extraOptions)
         allOptions.putAll(extraOptions);
 
         final String distributedBy =
-                Optional.ofNullable(distribution)
-                        .map(TableDistribution::asSerializableString)
-                        .orElse("");
+                Optional.ofNullable(distribution).map(TableDistribution::toString).orElse("");
         final String partitionedBy =
                 partitionKeys.isEmpty()
                         ? ""
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/TableDistribution.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/TableDistribution.java
index 8c0d44023961e..733bc1d765e97 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/TableDistribution.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/TableDistribution.java
@@ -116,7 +116,7 @@ public Optional<Integer> getBucketCount() {
         return Optional.ofNullable(bucketCount);
     }
 
-    public String asSerializableString() {
+    private String asSerializableString() {
         if (getBucketKeys().isEmpty()
                 && getBucketCount().isPresent()
                 && getBucketCount().get() != 0) {
@@ -142,4 +142,9 @@ && getBucketCount().get() != 0) {
         sb.append("\n");
         return sb.toString();
     }
+
+    @Override
+    public String toString() {
+        return asSerializableString();
+    }
 }