Skip to content

Commit 7e4675d

Browse files
zhicwuzhicwu
authored
Merge pull request #930 from zhicwu/grpc-patch
Fit & finish for patch9
2 parents 998c24b + 66cb081 commit 7e4675d

File tree

18 files changed

+102
-55
lines changed

18 files changed

+102
-55
lines changed

‎clickhouse-benchmark/src/main/java/com/clickhouse/benchmark/misc/StreamBenchmark.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -63,8 +63,7 @@ public static class StreamState extends BaseState {
6363

6464
@Setup(Level.Trial)
6565
public void setupSamples() {
66-
bufferSize = Integer.getInteger("buffer",
67-
(int) ClickHouseClientOption.WRITE_BUFFER_SIZE.getDefaultValue());
66+
bufferSize = Integer.getInteger("buffer", (int) ClickHouseClientOption.BUFFER_SIZE.getDefaultValue());
6867
samples = Integer.getInteger("samples", 500000);
6968

7069
bytes = new byte[samples];
@@ -74,7 +73,7 @@ public void setupSamples() {
7473
options.put(ClickHouseClientOption.REQUEST_BUFFERING, ClickHouseBufferingMode.valueOf(
7574
System.getProperty("mode", ClickHouseClientOption.REQUEST_BUFFERING.getDefaultValue().toString())
7675
.toUpperCase()));
77-
options.put(ClickHouseClientOption.WRITE_BUFFER_SIZE, bufferSize);
76+
options.put(ClickHouseClientOption.BUFFER_SIZE, bufferSize);
7877
options.put(ClickHouseClientOption.MAX_QUEUED_BUFFERS,
7978
Integer.getInteger("queue", (int) ClickHouseClientOption.MAX_QUEUED_BUFFERS.getDefaultValue()));
8079
options.put(ClickHouseClientOption.COMPRESS, Boolean.parseBoolean(System.getProperty("compress", "false")));

‎clickhouse-client/src/main/java/com/clickhouse/client/ClickHouseClient.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,7 @@ static ExecutorService getExecutorService() {
7676
static ClickHouseOutputStream getRequestOutputStream(ClickHouseConfig config, OutputStream output,
7777
Runnable postCloseAction) {
7878
if (config == null) {
79-
return ClickHouseOutputStream.of(output, (int) ClickHouseClientOption.WRITE_BUFFER_SIZE.getDefaultValue(),
79+
return ClickHouseOutputStream.of(output, (int) ClickHouseClientOption.BUFFER_SIZE.getDefaultValue(),
8080
ClickHouseCompression.NONE, postCloseAction);
8181
}
8282

@@ -151,7 +151,7 @@ static ClickHouseOutputStream getAsyncRequestOutputStream(ClickHouseConfig confi
151151
static ClickHouseInputStream getResponseInputStream(ClickHouseConfig config, InputStream input,
152152
Runnable postCloseAction) {
153153
if (config == null) {
154-
return ClickHouseInputStream.of(input, (int) ClickHouseClientOption.READ_BUFFER_SIZE.getDefaultValue(),
154+
return ClickHouseInputStream.of(input, (int) ClickHouseClientOption.BUFFER_SIZE.getDefaultValue(),
155155
ClickHouseCompression.NONE, postCloseAction);
156156
}
157157

‎clickhouse-client/src/main/java/com/clickhouse/client/ClickHouseConfig.java‎

Lines changed: 29 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -114,8 +114,10 @@ protected static final Object mergeMetricRegistry(List<ClickHouseConfig> list) {
114114
private final ClickHouseFormat format;
115115
private final int maxBufferSize;
116116
private final int bufferSize;
117+
private final int bufferQueueVariation;
117118
private final int readBufferSize;
118119
private final int writeBufferSize;
120+
private final int requestChunkSize;
119121
private final ClickHouseBufferingMode requestBuffering;
120122
private final ClickHouseBufferingMode responseBuffering;
121123
private final int maxExecutionTime;
@@ -198,12 +200,11 @@ public ClickHouseConfig(Map<ClickHouseOption, Serializable> options, ClickHouseC
198200
this.format = (ClickHouseFormat) getOption(ClickHouseClientOption.FORMAT, ClickHouseDefaults.FORMAT);
199201
this.maxBufferSize = ClickHouseUtils.getBufferSize((int) getOption(ClickHouseClientOption.MAX_BUFFER_SIZE), -1,
200202
-1);
201-
this.bufferSize = ClickHouseUtils.getBufferSize((int) getOption(ClickHouseClientOption.BUFFER_SIZE), -1,
202-
this.maxBufferSize);
203-
this.readBufferSize = ClickHouseUtils.getBufferSize(
204-
(int) getOption(ClickHouseClientOption.READ_BUFFER_SIZE), this.bufferSize, this.maxBufferSize);
205-
this.writeBufferSize = ClickHouseUtils.getBufferSize(
206-
(int) getOption(ClickHouseClientOption.WRITE_BUFFER_SIZE), this.bufferSize, this.maxBufferSize);
203+
this.bufferSize = (int) getOption(ClickHouseClientOption.BUFFER_SIZE);
204+
this.bufferQueueVariation = (int) getOption(ClickHouseClientOption.BUFFER_QUEUE_VARIATION);
205+
this.readBufferSize = (int) getOption(ClickHouseClientOption.READ_BUFFER_SIZE);
206+
this.writeBufferSize = (int) getOption(ClickHouseClientOption.WRITE_BUFFER_SIZE);
207+
this.requestChunkSize = (int) getOption(ClickHouseClientOption.REQUEST_CHUNK_SIZE);
207208
this.requestBuffering = (ClickHouseBufferingMode) getOption(ClickHouseClientOption.REQUEST_BUFFERING,
208209
ClickHouseDefaults.BUFFERING);
209210
this.responseBuffering = (ClickHouseBufferingMode) getOption(ClickHouseClientOption.RESPONSE_BUFFERING,
@@ -418,13 +419,24 @@ public int getBufferSize() {
418419
return bufferSize;
419420
}
420421

422+
/**
423+
* Gets number of times the buffer queue is filled up before
424+
* increasing capacity of buffer queue. Zero or negative value means the queue
425+
* length is fixed.
426+
*
427+
* @return variation
428+
*/
429+
public int getBufferQueueVariation() {
430+
return bufferQueueVariation;
431+
}
432+
421433
/**
422434
* Gets read buffer size in byte.
423435
*
424436
* @return read buffer size in byte
425437
*/
426438
public int getReadBufferSize() {
427-
return readBufferSize;
439+
return ClickHouseUtils.getBufferSize(readBufferSize, getBufferSize(), getMaxBufferSize());
428440
}
429441

430442
/**
@@ -433,7 +445,16 @@ public int getReadBufferSize() {
433445
* @return write buffer size in byte
434446
*/
435447
public int getWriteBufferSize() {
436-
return writeBufferSize;
448+
return ClickHouseUtils.getBufferSize(writeBufferSize, getBufferSize(), getMaxBufferSize());
449+
}
450+
451+
/**
452+
* Gets request chunk size.
453+
*
454+
* @return request chunk size
455+
*/
456+
public int getRequestChunkSize() {
457+
return ClickHouseUtils.getBufferSize(requestChunkSize, getWriteBufferSize(), getMaxBufferSize());
437458
}
438459

439460
/**

‎clickhouse-client/src/main/java/com/clickhouse/client/ClickHouseDataStreamFactory.java‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
import com.clickhouse.client.data.ClickHouseRowBinaryProcessor;
1212
import com.clickhouse.client.data.ClickHouseTabSeparatedProcessor;
1313
import com.clickhouse.client.stream.BlockingPipedOutputStream;
14+
import com.clickhouse.client.stream.CapacityPolicy;
1415
import com.clickhouse.client.stream.NonBlockingPipedOutputStream;
1516

1617
/**
@@ -44,7 +45,7 @@ public static ClickHouseDataStreamFactory getInstance() {
4445
* {@code input} is null
4546
* @param settings nullable settings
4647
* @param columns nullable list of columns
47-
* @return data processor
48+
* @return data processor, which might be null
4849
* @throws IOException when failed to read columns from input stream
4950
*/
5051
public ClickHouseDataProcessor getProcessor(ClickHouseConfig config, ClickHouseInputStream input,
@@ -131,7 +132,7 @@ public ClickHousePipedStream createPipedStream(ClickHouseConfig config) {
131132
return config != null
132133
? new ClickHousePipedStream(config.getWriteBufferSize(), config.getMaxQueuedBuffers(),
133134
config.getSocketTimeout())
134-
: new ClickHousePipedStream((int) ClickHouseClientOption.WRITE_BUFFER_SIZE.getDefaultValue(),
135+
: new ClickHousePipedStream((int) ClickHouseClientOption.BUFFER_SIZE.getDefaultValue(),
135136
(int) ClickHouseClientOption.MAX_QUEUED_BUFFERS.getDefaultValue(),
136137
(int) ClickHouseClientOption.SOCKET_TIMEOUT.getDefaultValue());
137138
}
@@ -146,27 +147,31 @@ public ClickHousePipedOutputStream createPipedOutputStream(ClickHouseConfig conf
146147
final int bufferSize = ClickHouseChecker.nonNull(config, "config").getWriteBufferSize();
147148
final boolean blocking;
148149
final int queue;
150+
final CapacityPolicy policy;
149151
final int timeout;
150152

151153
if (config.getResponseBuffering() == ClickHouseBufferingMode.PERFORMANCE) {
152154
blocking = false;
153155
queue = 0;
156+
policy = null;
154157
timeout = 0; // questionable
155158
} else {
156159
blocking = config.isUseBlockingQueue();
157160
queue = config.getMaxQueuedBuffers();
161+
policy = config.getBufferQueueVariation() < 1 ? CapacityPolicy.fixedCapacity(queue)
162+
: CapacityPolicy.linearDynamicCapacity(1, queue, config.getBufferQueueVariation());
158163
timeout = config.getSocketTimeout();
159164
}
160165
return blocking
161166
? new BlockingPipedOutputStream(bufferSize, queue, timeout, postCloseAction)
162-
: new NonBlockingPipedOutputStream(bufferSize, queue, timeout, null, postCloseAction);
167+
: new NonBlockingPipedOutputStream(bufferSize, queue, timeout, policy, postCloseAction);
163168
}
164169

165-
public ClickHousePipedOutputStream createPipedOutputStream(int writeBufferSize, int queueSize, int timeout,
170+
public ClickHousePipedOutputStream createPipedOutputStream(int bufferSize, int queueSize, int timeout,
166171
Runnable postCloseAction) {
167172
return new BlockingPipedOutputStream(
168-
ClickHouseUtils.getBufferSize(writeBufferSize,
169-
(int) ClickHouseClientOption.WRITE_BUFFER_SIZE.getDefaultValue(),
173+
ClickHouseUtils.getBufferSize(bufferSize,
174+
(int) ClickHouseClientOption.BUFFER_SIZE.getDefaultValue(),
170175
(int) ClickHouseClientOption.MAX_BUFFER_SIZE.getDefaultValue()),
171176
queueSize, timeout, postCloseAction);
172177
}

‎clickhouse-client/src/main/java/com/clickhouse/client/ClickHouseInputStream.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -244,7 +244,7 @@ public static ClickHouseInputStream of(InputStream... inputs) {
244244
if (inputs == null || inputs.length == 0) {
245245
return EmptyInputStream.INSTANCE;
246246
} else if (inputs.length == 1) {
247-
return of(inputs[0], (int) ClickHouseClientOption.READ_BUFFER_SIZE.getDefaultValue(), null, null);
247+
return of(inputs[0], (int) ClickHouseClientOption.BUFFER_SIZE.getDefaultValue(), null, null);
248248
}
249249
return of(Arrays.asList(inputs), InputStream.class, null, null);
250250
}
@@ -367,7 +367,7 @@ public static <T> ClickHouseInputStream of(Iterable<T> source, Class<T> clazz, F
367367
* @param input non-null input stream, which will be closed
368368
* @param output non-null output stream, which will remain open
369369
* @param bufferSize buffer size, zero or negative number will be treated as
370-
* {@link ClickHouseClientOption#WRITE_BUFFER_SIZE}
370+
* {@link ClickHouseClientOption#BUFFER_SIZE}
371371
* @return written bytes
372372
* @throws IOException when error occured reading from input stream or writing
373373
* data to output stream
@@ -378,7 +378,7 @@ public static long pipe(InputStream input, OutputStream output, int bufferSize)
378378
}
379379

380380
bufferSize = ClickHouseUtils.getBufferSize(bufferSize,
381-
(int) ClickHouseClientOption.WRITE_BUFFER_SIZE.getDefaultValue(),
381+
(int) ClickHouseClientOption.BUFFER_SIZE.getDefaultValue(),
382382
(int) ClickHouseClientOption.MAX_BUFFER_SIZE.getDefaultValue());
383383
return pipe(input, output, new byte[bufferSize]);
384384
}

‎clickhouse-client/src/main/java/com/clickhouse/client/ClickHouseOutputStream.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ public abstract class ClickHouseOutputStream extends OutputStream {
2929
* {@link ClickHouseOutputStream}
3030
*/
3131
public static ClickHouseOutputStream of(OutputStream output) {
32-
return of(output, (int) ClickHouseClientOption.WRITE_BUFFER_SIZE.getDefaultValue(), null, null);
32+
return of(output, (int) ClickHouseClientOption.BUFFER_SIZE.getDefaultValue(), null, null);
3333
}
3434

3535
/**

‎clickhouse-client/src/main/java/com/clickhouse/client/ClickHouseResponse.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -119,13 +119,13 @@ default ClickHouseRecord firstRecord() {
119119
*
120120
* @param output non-null output stream, which will remain open
121121
* @param bufferSize buffer size, 0 or negative value will be treated as
122-
* {@link ClickHouseClientOption#WRITE_BUFFER_SIZE}
122+
* {@link ClickHouseClientOption#BUFFER_SIZE}
123123
* @throws IOException when error occurred reading or writing data
124124
*/
125125
default void pipe(OutputStream output, int bufferSize) throws IOException {
126126
ClickHouseInputStream.pipe(getInputStream(), ClickHouseChecker.nonNull(output, "output"),
127127
ClickHouseUtils.getBufferSize(bufferSize,
128-
(int) ClickHouseClientOption.WRITE_BUFFER_SIZE.getDefaultValue(),
128+
(int) ClickHouseClientOption.BUFFER_SIZE.getDefaultValue(),
129129
(int) ClickHouseClientOption.MAX_BUFFER_SIZE.getDefaultValue()));
130130
}
131131

‎clickhouse-client/src/main/java/com/clickhouse/client/config/ClickHouseClientOption.java‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -24,26 +24,41 @@ public enum ClickHouseClientOption implements ClickHouseOption {
2424
* to {@link #MAX_BUFFER_SIZE} if it's too large.
2525
*/
2626
BUFFER_SIZE("buffer_size", 8192, "Default buffer size in byte for both request and response."),
27+
/**
28+
* Number of times the buffer queue is filled up before increasing capacity of
29+
* buffer queue. Zero or negative value means the queue length is fixed.
30+
*/
31+
BUFFER_QUEUE_VARIATION("buffer_queue_variation", 100,
32+
"Number of times the buffer queue is filled up before increasing capacity of buffer queue. Zero or negative value means the queue length is fixed."),
2733
/**
2834
* Read buffer size in byte. It's mainly for input stream(e.g. reading data from
2935
* server response). Its value defaults to {@link #BUFFER_SIZE}, and it will be
3036
* reset to {@link #MAX_BUFFER_SIZE} when it's too large.
3137
*/
32-
READ_BUFFER_SIZE("read_buffer_size", BUFFER_SIZE.getDefaultValue(), "Read buffer size in byte"),
38+
READ_BUFFER_SIZE("read_buffer_size", 0,
39+
"Read buffer size in byte, zero or negative value means same as buffer_size"),
3340
/**
3441
* Write buffer size in byte. It's mainly for output stream(e.g. writing data
3542
* into request). Its value defaults to {@link #BUFFER_SIZE}, and it will
3643
* be reset to {@link #MAX_BUFFER_SIZE} when it's too large.
3744
*/
38-
WRITE_BUFFER_SIZE("write_buffer_size", BUFFER_SIZE.getDefaultValue(), "Write buffer size in byte"),
45+
WRITE_BUFFER_SIZE("write_buffer_size", 0,
46+
"Write buffer size in byte, zero or negative value means same as buffer_size"),
47+
/**
48+
* Maximum request chunk size in byte.
49+
*/
50+
REQUEST_CHUNK_SIZE("request_chunk_size", 0,
51+
"Maximum request chunk size in byte, zero or negative value means same as write_buffer_size"),
3952
/**
4053
* Request buffering mode.
4154
*/
42-
REQUEST_BUFFERING("request_buffering", ClickHouseDefaults.BUFFERING, "Request buffering mode"),
55+
REQUEST_BUFFERING("request_buffering", ClickHouseDefaults.BUFFERING.getDefaultValue(),
56+
"Request buffering mode"),
4357
/**
4458
* Response buffering mode.
4559
*/
46-
RESPONSE_BUFFERING("response_buffering", ClickHouseDefaults.BUFFERING, "Response buffering mode."),
60+
RESPONSE_BUFFERING("response_buffering", ClickHouseDefaults.BUFFERING.getDefaultValue(),
61+
"Response buffering mode."),
4762
/**
4863
* Client name.
4964
*/

‎clickhouse-client/src/main/java/com/clickhouse/client/data/ClickHousePipedStream.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ public ClickHousePipedStream(int bufferSize, int queueLength, int timeout) {
4343

4444
// may need an initialBufferSize and a monitor to update bufferSize in runtime
4545
this.bufferSize = ClickHouseUtils.getBufferSize(bufferSize,
46-
(int) ClickHouseClientOption.WRITE_BUFFER_SIZE.getDefaultValue(),
46+
(int) ClickHouseClientOption.BUFFER_SIZE.getDefaultValue(),
4747
(int) ClickHouseClientOption.MAX_BUFFER_SIZE.getDefaultValue());
4848
this.timeout = timeout;
4949

@@ -103,7 +103,10 @@ public void close() throws IOException {
103103
Thread.currentThread().interrupt();
104104
throw new IOException("Thread was interrupted when putting EMPTY buffer into queue", e);
105105
} finally {
106-
super.close();
106+
closed = true;
107+
if (postCloseAction != null) {
108+
postCloseAction.run();
109+
}
107110
}
108111
}
109112

‎clickhouse-client/src/main/java/com/clickhouse/client/stream/BlockingPipedOutputStream.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ public BlockingPipedOutputStream(int bufferSize, int queueLength, int timeout, R
3838

3939
// may need an initialBufferSize and a monitor to update bufferSize in runtime
4040
this.bufferSize = ClickHouseUtils.getBufferSize(bufferSize,
41-
(int) ClickHouseClientOption.WRITE_BUFFER_SIZE.getDefaultValue(),
41+
(int) ClickHouseClientOption.BUFFER_SIZE.getDefaultValue(),
4242
(int) ClickHouseClientOption.MAX_BUFFER_SIZE.getDefaultValue());
4343
this.timeout = timeout;
4444

0 commit comments

Comments
 (0)