Skip to content

Commit 0dc8ad6

Browse files
zgdgodzgdgod
committed
holo-client更新至2.4.0，connector发布1.4.1版本
1 parent fad463d commit 0dc8ad6

35 files changed

Lines changed: 958 additions & 455 deletions

‎holo-client/README.md‎

Lines changed: 22 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -554,6 +554,16 @@ unnest格式相比multi values有如下优点:
554554
- 分区表drop column之后，自动创建子分区时，可能拿到错误的分区值，bug引入版本1.X，bug修复版本2.2.11
555555
- 当用户表名中有下划线时，可能获取到错误的字段信息，bug引入版本1.X，bug修复版本2.2.12
556556

557+
## Release Note
558+
- 2.4.0
559+
- 新建连接时会设置LOGIN_TIMEOUT为60S，之前未设置
560+
- 对实例只读状态的报错增加重试
561+
- 修复binlog读取timestamp类型微秒精度丢失
562+
- table schema发生变化时,会强制flush
563+
- 支持设置连接最大存活时间，默认为24h
564+
- 支持激进模式，开启后如果连接空闲，不等攒批满就触发flush，在流量小时可以降低写入延迟
565+
- 查询非utf8字符不抛出异常
566+
557567
## 附录
558568
### HoloConfig参数说明
559569
#### 基础配置
@@ -593,8 +603,9 @@ unnest格式相比multi values有如下优点:
593603
| maxRowsPerSql | Integer.MAX_VALUE | useLegacyPutHandler=false，且通过unnest形式写入时，每条sql的最大行数 | 2.0.1 |
594604
| maxBytesPerSql | Long.MAX_VALUE | useLegacyPutHandler=false，且通过unnest形式写入时，每条sql的最大字节数 | 2.0.1 |
595605
| enableAffectedRows | false | 开启时 若用户用holoclient.sql执行statement.executeUpdate将会返回正确的affectrow计数，但对于行存表进行holoclient.put会有性能下降 | 2.2.5 |
596-
| enableGenerateBinlog | true | 关闭时，通过当前holo-client写入的数据不会生成binlog | 2.2.11 |
597-
| enableDeduplication | true | 写入时是否对攒批数据做去重，设置为false表示不会去重，如果数据重复非常严重，性能最差相当于writeBatchSize设置为1的逐条写入. | 2.3.0 |
606+
| enableGenerateBinlog | true | 关闭时，通过当前holo-client写入的数据不会生成binlog | 2.2.11 |
607+
| enableDeduplication | true | 写入时是否对攒批数据做去重，设置为false表示不会去重，如果数据重复非常严重，性能最差相当于writeBatchSize设置为1的逐条写入. | 2.3.0 |
608+
| enableAggressive | false | 写入激进模式，开启后如果连接空闲，不等攒批满就会触发flush，在流量小时可以降低写入延迟. | 2.4.0 |
598609

599610
#### 查询配置
600611
| 参数名 | 默认值 | 说明 |引入版本|
@@ -608,14 +619,15 @@ unnest格式相比multi values有如下优点:
608619
| readRetryCount | 1 | Get操作的尝试次数，1表示不重试 | 2.1.5 |
609620

610621
#### 连接配置
611-
| 参数名 | 默认值 | 说明 |引入版本|
612-
| --- | --- | --- | --- |
613-
| retryCount | 3 | 当连接故障时，写入和查询的重试次数 | 1.2.3|
614-
| retrySleepInitMs | 1000 | 每次重试的等待时间=retrySleepInitMs+retry*retrySleepStepMs | 1.2.3 |
615-
| retrySleepStepMs | 10000 | 每次重试的等待时间=retrySleepInitMs+retry*retrySleepStepMs |1.2.3 |
616-
| connectionMaxIdleMs| 60000 | 写入线程和点查线程数据库连接的最大Idle时间，超过连接将被释放| 1.2.4 |
617-
| metaCacheTTL | 1 min | getTableSchema信息的本地缓存时间 | 1.2.6 |
618-
| metaAutoRefreshFactor | 4 | 当tableSchema cache剩余存活时间短于 metaCacheTTL/metaAutoRefreshFactor 将自动刷新cache | 1.2.10.1 |
622+
| 参数名 | 默认值 | 说明 | 引入版本 |
623+
| --- |--------------------------------|--------------------------------------------------------------------------|----------|
624+
| retryCount | 3 | 当连接故障时，写入和查询的重试次数 | 1.2.3 |
625+
| retrySleepInitMs | 1000 | 每次重试的等待时间=retrySleepInitMs+retry*retrySleepStepMs | 1.2.3 |
626+
| retrySleepStepMs | 10000 | 每次重试的等待时间=retrySleepInitMs+retry*retrySleepStepMs | 1.2.3 |
627+
| connectionMaxIdleMs| 60000 | 写入线程和点查线程数据库连接的最大Idle时间，超过连接将在空闲时被释放，使用时重建 | 1.2.4 |
628+
| connectionMaxAliveMs| 86400000(24 * 60 * 60 * 1000L) | 写入线程和点查线程数据库连接的最大存活时间，超过连接将在空闲时被释放，使用时重建 | 2.4.0 |
629+
| metaCacheTTL | 1 min | getTableSchema信息的本地缓存时间 | 1.2.6 |
630+
| metaAutoRefreshFactor | 4 | 当tableSchema cache剩余存活时间短于 metaCacheTTL/metaAutoRefreshFactor 将自动刷新cache | 1.2.10.1 |
619631

620632
#### 消费Binlog配置
621633
| 参数名 | 默认值 | 说明 |引入版本 |

‎holo-client/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66

77
<groupId>com.alibaba.hologres</groupId>
88
<artifactId>holo-client</artifactId>
9-
<version>2.3.0</version>
9+
<version>2.4.0</version>
1010

1111
<url>https://www.hologres.io/</url>
1212
<name>holo-client</name>

‎holo-client/src/main/java/com/alibaba/hologres/client/HoloClient.java‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -666,14 +666,15 @@ public void setAsyncCommit(boolean asyncCommit) {
666666
this.asyncCommit = asyncCommit;
667667
}
668668

669-
private void closeInternal() {
670-
669+
private void closeInternal() throws HoloClientException{
670+
HoloClientException exception = null;
671671
if (pool != null && pool.isRegister(this)) {
672672
try {
673673
tryThrowException();
674674
flush();
675675
} catch (HoloClientException e) {
676676
LOGGER.error("fail when close", e);
677+
exception = e;
677678
}
678679
pool.unregister(this);
679680
if (isEmbeddedPool) {
@@ -686,10 +687,17 @@ private void closeInternal() {
686687
fixedPool.close();
687688
}
688689
}
690+
if (null != exception) {
691+
throw exception;
692+
}
689693
}
690694

691695
@Override
692696
public void close() {
693-
closeInternal();
697+
try {
698+
closeInternal();
699+
} catch (HoloClientException e) {
700+
throw new RuntimeException(e);
701+
}
694702
}
695703
}

‎holo-client/src/main/java/com/alibaba/hologres/client/HoloConfig.java‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -245,6 +245,14 @@ public class HoloConfig implements Serializable {
245245
*/
246246
boolean enableDeduplication = true;
247247

248+
/**
249+
* 激进模式,写入时发现连接空闲,即使未攒够批也立刻提交.
250+
* boolean
251+
*
252+
* @HasGetter
253+
* @HasSetter
254+
*/
255+
boolean enableAggressive = false;
248256
//--------------------------read conf-------------------------------------------------
249257
/**
250258
* 最多一次将readBatchSize条Get请求合并提交，默认128.
@@ -373,6 +381,15 @@ public class HoloConfig implements Serializable {
373381
*/
374382
long connectionMaxIdleMs = 60000L;
375383

384+
/**
385+
* 每个get和put的后台连接最大存活时间,超过之后将被释放(再次使用时会自动重新连接).
386+
* 尽可能避免服务端可能存在的内存泄漏等问题.
387+
*
388+
* @HasGetter
389+
* @HasSetter
390+
*/
391+
long connectionMaxAliveMs = 24 * 60 * 60 * 1000L;
392+
376393
/**
377394
* meta信息缓存时间(ms).
378395
*
@@ -647,6 +664,14 @@ public void setConnectionMaxIdleMs(long connectionMaxIdleMs) {
647664
this.connectionMaxIdleMs = connectionMaxIdleMs;
648665
}
649666

667+
public long getConnectionMaxAliveMs() {
668+
return connectionMaxAliveMs;
669+
}
670+
671+
public void setConnectionMaxAliveMs(long connectionMaxAliveMs) {
672+
this.connectionMaxAliveMs = connectionMaxAliveMs;
673+
}
674+
650675
public int getWriteThreadSize() {
651676
return writeThreadSize;
652677
}
@@ -919,6 +944,14 @@ public void setEnableDeduplication(boolean enableDedup) {
919944
this.enableDeduplication = enableDedup;
920945
}
921946

947+
public boolean isEnableAggressive() {
948+
return enableAggressive;
949+
}
950+
951+
public void setEnableAggressive(boolean enableAggressive) {
952+
this.enableAggressive = enableAggressive;
953+
}
954+
922955
public static String[] getPropertyKeys() {
923956
Field[] fields = HoloConfig.class.getDeclaredFields();
924957
String[] propertyKeys = new String[fields.length];

‎holo-client/src/main/java/com/alibaba/hologres/client/copy/RecordBinaryOutputStream.java‎

Lines changed: 17 additions & 77 deletions
Original file line numberDiff line numberDiff line change
@@ -9,21 +9,16 @@
99
import com.alibaba.hologres.client.model.TableSchema;
1010
import org.postgresql.core.BaseConnection;
1111
import org.postgresql.jdbc.ArrayUtil;
12+
import org.postgresql.jdbc.TimestampUtil;
1213

1314
import java.io.IOException;
1415
import java.io.OutputStream;
1516
import java.math.BigDecimal;
1617
import java.math.RoundingMode;
18+
import java.sql.Array;
1719
import java.sql.Date;
1820
import java.sql.SQLException;
19-
import java.sql.Timestamp;
2021
import java.sql.Types;
21-
import java.time.OffsetDateTime;
22-
import java.time.format.DateTimeFormatter;
23-
import java.time.format.DateTimeFormatterBuilder;
24-
import java.time.temporal.ChronoField;
25-
import java.util.TimeZone;
26-
import java.util.concurrent.TimeUnit;
2722

2823
/**
2924
* Record转pg binary流.
@@ -46,6 +41,16 @@ private void fillHeader() throws IOException {
4641
writeInt(0);
4742
}
4843

44+
@Override
45+
public void close() throws IOException {
46+
if (!fillHeader) {
47+
fillHeader = true;
48+
fillHeader();
49+
writeCellBuffer();
50+
}
51+
super.close();
52+
}
53+
4954
@Override
5055
protected void fillByteBuffer(Record record) throws IOException {
5156
if (!fillHeader) {
@@ -183,50 +188,8 @@ private void fillByteBuffer(Object obj, Column column)
183188
}
184189
break;
185190
case Types.TIMESTAMP:
186-
if (obj instanceof Timestamp) {
187-
Timestamp ts = (Timestamp) obj;
188-
long seconds = javaEpochToPg(ts.getTime() / 1000, TimeUnit.SECONDS);
189-
// Convert to micros rounding nanoseconds
190-
long micros =
191-
TimeUnit.SECONDS.toMicros(seconds)
192-
+ TimeUnit.NANOSECONDS.toMicros(ts.getNanos() + 500);
193-
if ("timestamp".equals(typeName)) {
194-
micros += TimeZone.getDefault().getRawOffset() * 1000L;
195-
}
196-
writeInt(8);
197-
writeLong(micros);
198-
} else if (obj instanceof String) {
199-
OffsetDateTime dateTime =
200-
OffsetDateTime.parse((String) obj, DATE_TIME_FORMATTER);
201-
long seconds = javaEpochToPg(dateTime.toEpochSecond(), TimeUnit.SECONDS);
202-
// Convert to micros rounding nanoseconds
203-
long micros =
204-
TimeUnit.SECONDS.toMicros(seconds)
205-
+ TimeUnit.NANOSECONDS.toMicros(dateTime.getNano() + 500);
206-
writeInt(8);
207-
writeLong(micros);
208-
} else if (obj instanceof Number) {
209-
long ms = ((Number) obj).longValue();
210-
long seconds = javaEpochToPg(ms / 1000L, TimeUnit.SECONDS);
211-
// Convert to micros rounding nanoseconds
212-
long micros =
213-
TimeUnit.SECONDS.toMicros(seconds)
214-
+ TimeUnit.NANOSECONDS.toMicros((ms % 1000) * 1000000L + 500);
215-
writeInt(8);
216-
writeLong(micros);
217-
} else if (obj instanceof java.util.Date) {
218-
long ms = ((java.util.Date) obj).getTime();
219-
long seconds = javaEpochToPg(ms / 1000L, TimeUnit.SECONDS);
220-
// Convert to micros rounding nanoseconds
221-
long micros =
222-
TimeUnit.SECONDS.toMicros(seconds)
223-
+ TimeUnit.NANOSECONDS.toMicros((ms % 1000) * 1000000L + 500);
224-
writeInt(8);
225-
writeLong(micros);
226-
} else {
227-
throw new RuntimeException(
228-
"unsupported type for timestamp " + obj.getClass().getName());
229-
}
191+
writeInt(8);
192+
writeLong(TimestampUtil.timestampToPgEpochMicroSecond(obj, typeName));
230193
break;
231194
case Types.BINARY:
232195
if (obj instanceof byte[]) {
@@ -296,7 +259,9 @@ private void fillByteBuffer(Object obj, Column column)
296259
throw new IOException("unsupported type:" + typeName + "(" + type + "). Please call RecordBinaryOutputSteam constructor with BaseConnection Param");
297260
}
298261
try {
299-
byte[] arrayBytes = ArrayUtil.arrayToBinary(conn, obj, column.getTypeName());
262+
// obj如果是List<>或Object[]，都尝试转成Array
263+
Array array = ArrayUtil.objectToArray(conn, obj, column.getTypeName());
264+
byte[] arrayBytes = ArrayUtil.arrayToBinary(conn, array != null ? ArrayUtil.objectToArray(conn, obj, column.getTypeName()) : obj, column.getTypeName());
300265
writeInt(arrayBytes.length);
301266
write(arrayBytes);
302267
} catch (SQLException e) {
@@ -387,29 +352,4 @@ private static short[] encodeFromString(String num, short[] info) {
387352
info[2] = displayScale;
388353
return digits;
389354
}
390-
391-
public static final DateTimeFormatter DATE_TIME_FORMATTER =
392-
new DateTimeFormatterBuilder()
393-
.optionalStart()
394-
.append(DateTimeFormatter.ISO_LOCAL_DATE)
395-
.optionalEnd()
396-
.optionalStart()
397-
.appendLiteral(' ')
398-
.optionalEnd()
399-
.optionalStart()
400-
.appendLiteral('T')
401-
.optionalEnd()
402-
.appendPattern("HH:mm:ss")
403-
.appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
404-
.optionalStart()
405-
.appendOffset("+HH", "Z")
406-
.optionalEnd()
407-
.toFormatter();
408-
409-
private static final long PG_EPOCH_SECS = 946684800L;
410-
411-
static long javaEpochToPg(long value, TimeUnit timeUnit) {
412-
413-
return value - timeUnit.convert(PG_EPOCH_SECS, TimeUnit.SECONDS);
414-
}
415355
}

‎holo-client/src/main/java/com/alibaba/hologres/client/copy/RecordOutputStream.java‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,9 +64,12 @@ public void putRecord(Record record) throws IOException {
6464
throw new IOException("RecordOutputFormat already closed");
6565
}
6666
fillByteBuffer(record);
67+
writeCellBuffer();
68+
}
69+
70+
protected void writeCellBuffer() throws IOException {
6771
cellBuffer.flip();
6872
os.write(cellBuffer.array(), cellBuffer.position(), cellBuffer.remaining());
69-
7073
cellBuffer.clear();
7174
}
7275

‎holo-client/src/main/java/com/alibaba/hologres/client/copy/RecordTextOutputStream.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
import java.io.OutputStream;
1818
import java.io.StringWriter;
1919
import java.nio.charset.Charset;
20+
import java.sql.Array;
2021
import java.sql.Types;
2122

2223
import static com.alibaba.hologres.client.copy.CopyUtil.ESCAPE;
@@ -56,7 +57,8 @@ protected void fillByteBuffer(Record record) throws IOException {
5657
case Types.ARRAY:
5758
String text;
5859
if (obj.getClass().getComponentType() != null) {
59-
text = ArrayUtil.arrayToString(obj);
60+
Array array = ArrayUtil.objectToArray(conn, obj, column.getTypeName());
61+
text = ArrayUtil.arrayToString(array != null ? ArrayUtil.objectToArray(conn, obj, column.getTypeName()) : obj);
6062
} else {
6163
text = String.valueOf(obj);
6264
}

‎holo-client/src/main/java/com/alibaba/hologres/client/exception/HoloClientException.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,10 @@ public static HoloClientException fromSqlException(SQLException e) {
3636
code = ExceptionCode.CONNECTION_ERROR;
3737
} else if (e.getMessage() != null && e.getMessage().contains("not allowed in readonly mode")) {
3838
code = ExceptionCode.READ_ONLY;
39-
} else if (e.getMessage() != null && e.getMessage().contains("Resource busy")) {
39+
} else if (e.getMessage() != null && (e.getMessage().contains("Resource busy")
40+
|| e.getMessage().contains("Fail to fetch table group meta from store master")
41+
|| e.getMessage().contains("Get rundown is not allowed in recovering state")
42+
|| e.getMessage().contains("the workers or shards are unhealthy"))) {
4043
code = ExceptionCode.BUSY;
4144
} else if (e.getMessage() != null && (e.getMessage().contains("too many clients already") || e.getMessage().contains("remaining connection slots are reserved"))) {
4245
code = ExceptionCode.TOO_MANY_CONNECTIONS;
@@ -53,7 +56,7 @@ public static HoloClientException fromSqlException(SQLException e) {
5356
} else if (PSQLState.SYNTAX_ERROR.getState().equals(state)) {
5457
code = ExceptionCode.SYNTAX_ERROR;
5558
} else if (PSQLState.UNDEFINED_COLUMN.getState().equals(state) || (e.getMessage() != null && (e.getMessage().contains("Invalid table id") || e.getMessage().contains("Refresh meta timeout") ||
56-
e.getMessage().contains("mismatches the version of the table") || e.getMessage().contains("could not open relation with OID") || e.getMessage().contains("replay not finished yet")))) {
59+
e.getMessage().contains("mismatches the version of the table") || e.getMessage().contains("could not open relation with OID") || e.getMessage().contains("replay not finished yet") || e.getMessage().contains("fail to execute query Table not found")))) {
5760
//大量删分区的时, 查表分区是否存在 会报could not open relation with OID
5861
//Invalid table id , SQLState = UNDEFINED_TABLE
5962
//Check META_NOT_MATCH First.

0 commit comments

Comments
 (0)