Add DefaultRecordBatch implementation aka V2 message format parser/builder. (#1185)

Added bytecode optimization for varint and append/read_msg functions. Mostly based on avoiding LOAD_GLOBAL calls.

Author: tvoinarovskyi

benchmarks/record_batch_compose.py

@@ -58,7 +58,8 @@ def func(loops, magic):
             magic, batch_size=DEFAULT_BATCH_SIZE, compression_type=0)
         for _ in range(MESSAGES_PER_BATCH):
             key, value, timestamp = next(precomputed_samples)
-            size = batch.append(timestamp=timestamp, key=key, value=value)
+            size = batch.append(
+                timestamp=timestamp, key=key, value=value)
             assert size
         batch.close()
         results.append(batch.buffer())
@@ -73,3 +74,4 @@ def func(loops, magic):
 runner = perf.Runner()
 runner.bench_time_func('batch_append_v0', func, 0)
 runner.bench_time_func('batch_append_v1', func, 1)
+runner.bench_time_func('batch_append_v2', func, 2)

benchmarks/record_batch_read.py

@@ -35,7 +35,8 @@ def prepare(magic):
             size = batch.append(
                 random.randint(*TIMESTAMP_RANGE),
                 random_bytes(KEY_SIZE),
-                random_bytes(VALUE_SIZE))
+                random_bytes(VALUE_SIZE),
+                headers=[])
             assert size
         batch.close()
         samples.append(bytes(batch.buffer()))
@@ -78,3 +79,4 @@ def func(loops, magic):
 runner = perf.Runner()
 runner.bench_time_func('batch_read_v0', func, 0)
 runner.bench_time_func('batch_read_v1', func, 1)
+runner.bench_time_func('batch_read_v2', func, 2)