Add support array map tuple

[mzitnik]

Summary

Closes #57

Checklist

Delete items not relevant to your PR:

• Closes Add support for Array, Map,Tuple #57
• Unit and integration tests covering the common scenarios were added
• A human-readable description of the changes was provided to include in CHANGELOG
• For significant changes, documentation in https://github.com/ClickHouse/clickhouse-docs was updated with further explanations or tutorials

[windsurf-bot]

Other comments (9)

• flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/test/embedded/clickhouse/ClickHouseServerForTests.java (106-106) The SQL query uses incorrect syntax for table names. In ClickHouse, table names should be enclosed in backticks, not single quotes.

          String showDataSql = String.format("select * from `%s`", tableName);
  

• flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/test/embedded/clickhouse/ClickHouseServerForTests.java (107-107) The client connection is not closed after use, which could lead to resource leaks. Consider adding a try-with-resources block or explicitly closing the client after use.
• flink-connector-clickhouse-2.0.0/src/test/java/org/apache/flink/connector/clickhouse/sink/pojo/SimplePOJO.java (146-149) Consider using an immutable list for `tupleOfObjects` to better represent the immutable nature of tuples. You can use `Collections.unmodifiableList()` or similar approaches to prevent modifications after creation:

      this.tupleOfObjects = Collections.unmodifiableList(Arrays.asList("test", 1L, true));
  

• flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/pojo/SimplePOJO.java (72-72) Consider using an immutable collection for the tupleOfObjects field to better represent the immutable nature of tuples. You could use Collections.unmodifiableList() or a dedicated tuple class if one is available in your project.
• flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/pojo/SimplePOJO.java (252-258) The getter methods for collections (getStringList, getLongList, getMapOfStrings, getTupleOfObjects) return direct references to the mutable collections. Consider returning defensive copies or unmodifiable views to prevent external code from modifying the internal state of this object.

      public List<String> getStringList() { return Collections.unmodifiableList(stringList); }
  
      public List<Long> getLongList() { return Collections.unmodifiableList(longList); }
  
      public Map<String, String> getMapOfStrings() { return Collections.unmodifiableMap(mapOfStrings); }
  
      public List<Object> getTupleOfObjects() { return Collections.unmodifiableList(tupleOfObjects); }
  

• flink-connector-clickhouse-2.0.0/src/test/java/org/apache/flink/connector/clickhouse/sink/ClickHouseSinkTests.java (49-52) The string manipulation in this method is fragile. If the SQL string format changes, removing the last character might not work as expected.

  private String createSimplePOJOTableSQL(String database, String tableName, int parts_to_throw_insert) {
      String createTable = createSimplePOJOTableSQL(database, tableName);
      // Remove trailing semicolon if present and add settings
      if (createTable.trim().endsWith(";")) {
          return createTable.trim().substring(0, createTable.trim().length() - 1) + " " + 
                 String.format("SETTINGS parts_to_throw_insert = %d;", parts_to_throw_insert);
      } else {
          return createTable.trim() + " " + 
                 String.format("SETTINGS parts_to_throw_insert = %d;", parts_to_throw_insert);
      }
  }
  

• flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/ClickHouseSinkTests.java (290-290) These debug statements should be removed before merging. If they're useful for development, consider adding them behind a debug flag or removing them completely.
• flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/clickhouse/sink/ClickHouseSinkTests.java (540-540) Another debug statement that should be removed before merging.
• flink-connector-clickhouse-1.17/src/test/java/org/apache/flink/connector/test/embedded/clickhouse/ClickHouseServerForTests.java (110-113) Consider using the class logger (LOG) instead of System.out.println for consistency with the rest of the class.

💡 To request another review, post a new comment with "/windsurf-review".

[windsurf-bot]

There's a copy-paste error in the field names for UInt16, UInt32, and UInt64 serialization. They're all using 'uint8Primitive' and 'uint8Object' instead of their respective field names.

For example:

Suggested change

	Serialize.writeUInt16(out, input.getUint16Primitive(), false, false, ClickHouseDataType.UInt16, false, "uint8Primitive");
	Serialize.writeUInt16(out, input.getUint16Object(), false, false, ClickHouseDataType.UInt16, false, "uint8Object");
	Serialize.writeUInt32(out, input.getUint32Primitive(), false, false, ClickHouseDataType.UInt32, false, "uint8Primitive");
	Serialize.writeUInt32(out, input.getUint32Object(), false, false, ClickHouseDataType.UInt32, false, "uint8Object");
	Serialize.writeUInt64(out, input.getUint64Primitive(), false, false, ClickHouseDataType.UInt64, false, "uint8Primitive");
	Serialize.writeUInt64(out, input.getUint64Object(), false, false, ClickHouseDataType.UInt64, false, "uint8Object");
	Serialize.writeUInt16(out, input.getUint16Primitive(), false, false, ClickHouseDataType.UInt16, false, "uint16Primitive");
	Serialize.writeUInt16(out, input.getUint16Object(), false, false, ClickHouseDataType.UInt16, false, "uint16Object");

Similar corrections are needed for UInt32 and UInt64 calls.

[windsurf-bot]

The field names for UInt128 and UInt256 serialization are incorrect. They're using 'bigInteger128' and 'bigInteger256' which are likely the field names for the signed integer types.

Suggested change

	Serialize.writeUInt128(out, input.getUint128Object(), false, false, ClickHouseDataType.UInt128, false, "bigInteger128");
	Serialize.writeUInt256(out, input.getUint256Object(), false, false, ClickHouseDataType.UInt256, false, "bigInteger256");
	Serialize.writeUInt128(out, input.getUint128Object(), false, false, ClickHouseDataType.UInt128, false, "uint128Object");
	Serialize.writeUInt256(out, input.getUint256Object(), false, false, ClickHouseDataType.UInt256, false, "uint256Object");

[windsurf-bot]

The substring operation here could throw StringIndexOutOfBoundsException if the resulting string is less than 10 characters. Consider adding a check or padding the string to ensure it's at least 10 characters long.