Skip to content

Commit

Permalink
Merge pull request apache#11 from pabloem/basic-coders-progress-1
Browse files Browse the repository at this point in the history 
progress on basic coders
  • Loading branch information
@pabloem
pabloem authored Jan 7, 2022
2 parents 3e23533 + b226bdc commit 884ad82
Show file tree
Hide file tree
Showing 3 changed files with 107 additions and 21 deletions.
6 changes: 3 additions & 3 deletions sdks/node-ts/src/apache_beam/coders/coders.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,12 @@ interface Class<T> {

class CoderRegistry {
internal_registry = {};
get(urn: string): Coder<any> {
get(urn: string): Class<Coder<any>> {
const constructor: Class<Coder<any>> = this.internal_registry[urn];
if (constructor === undefined) {
return null!;
throw new Error('Could not find coder for URN ' + urn)
}
return new constructor();
return constructor;
}

register(urn: string, coderClass: Class<Coder<any>>) {
Expand Down
109 changes: 94 additions & 15 deletions sdks/node-ts/src/apache_beam/coders/standard_coders.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import * as runnerApi from '../proto/beam_runner_api';
import * as translations from '../internal/translations'

import { Writer, Reader } from 'protobufjs';
import struct = require('python-struct');
import { Coder, Context, CODER_REGISTRY } from "./coders";


Expand All @@ -17,16 +18,20 @@ class FakeCoder<T> implements Coder<T> {

export class BytesCoder implements Coder<Uint8Array> {
static URN: string = "beam:coder:bytes:v1";
static INSTANCE: BytesCoder = new BytesCoder();
type: string = "bytescoder";

encode(value: Uint8Array, writer: Writer, context: Context) {
var writeBytes =
function writeBytes_for(val, buf, pos) {
for (var i = 0; i < val.length; ++i)
for (var i = 0; i < val.length; ++i){
buf[pos + i] = val[i];
}

};

var len = value.length;
var hackedWriter = <any> writer;
var hackedWriter = <any>writer;
switch (context) {
case Context.wholeStream:
hackedWriter._push(writeBytes, len, value);
Expand Down Expand Up @@ -72,7 +77,7 @@ export type KV<K, V> = {
}

export class KVCoder<K, V> extends FakeCoder<KV<K, V>> {
static URN: string = "beam:coder:kvcoder:v1";
static URN: string = "beam:coder:kv:v1";
type: string = 'kvcoder';

keyCoder: Coder<K>;
Expand All @@ -95,6 +100,27 @@ export class KVCoder<K, V> extends FakeCoder<KV<K, V>> {
],
}
}

encode(element: KV<K, V>, writer: Writer, context: Context) {
console.log('encoding key ' + element.key as string + ' and value ' + element.value as string)
this.keyCoder.encode(element.key, writer, context);
this.valueCoder.encode(element.value, writer, context);
}

decode(reader: Reader, context: Context): KV<K, V> {
// const encodedKey = BytesCoder.INSTANCE.decode(reader, context);
// const encodedValue = BytesCoder.INSTANCE.decode(reader, context);
console.log((this.keyCoder as any));
console.log((this.valueCoder as any));
// return {
// 'key': this.keyCoder.decode(new Reader(encodedKey), Context.wholeStream),
// 'value': this.valueCoder.decode(new Reader(encodedValue), context)
// };
return {
'key': this.keyCoder.decode(reader, context),
'value': this.valueCoder.decode(reader, context)
}
}
}
CODER_REGISTRY.register(KVCoder.URN, KVCoder);

Expand All @@ -121,35 +147,87 @@ export class IterableCoder<T> extends FakeCoder<Iterable<T>> {
}
CODER_REGISTRY.register(IterableCoder.URN, IterableCoder);

export class StrUtf8Coder extends FakeCoder<String> {
export class StrUtf8Coder implements Coder<String> {
static URN: string = "beam:coder:string_utf8:v1";
type: string = 'stringutf8coder';
encoder = new TextEncoder();
decoder = new TextDecoder();

constructor() {
super();
}

encode(element: String, writer: Writer, context: Context) {
writer.bytes(this.encoder.encode(element as string));
const encodedElement = this.encoder.encode(element as string);
BytesCoder.INSTANCE.encode(encodedElement, writer, context);
}

decode(reader: Reader, context: Context): String {
return this.decoder.decode(reader.bytes());
return this.decoder.decode(BytesCoder.INSTANCE.decode(reader, context));
}
}
CODER_REGISTRY.register(StrUtf8Coder.URN, StrUtf8Coder);

export class VarIntCoder extends FakeCoder<Long | Number | BigInt> {

// TODO(pabloem): Is this the most efficient implementation?
export class VarIntCoder implements Coder<Long | Number | BigInt> {
static URN: string = "beam:coder:varint:v1";
type: string = "varintcoder";
encode(element: Number | Long | BigInt, writer: Writer, context: Context) {
writer.uint64(element as number);
var numEl = element as number
var encoded = [];
if (numEl < 0) {
// TODO(pabloem): Unable to encode negative integers due to JS encoding them
// internally as floats. We need to change this somewhat.
numEl += (1 << 64)
if (numEl <= 0) {
throw new Error('Value too large (negative (' + numEl + ')).')
}
}
while (true) {
var bits = numEl & 0x7F
numEl >>= 7
if (numEl) {
bits |= 0x80
}
encoded.push(bits as never);
if (!numEl) {
break
}
}

const encodedArr = new Uint8Array(encoded);
BytesCoder.INSTANCE.encode(encodedArr, writer, context);
}

decode(reader: Reader, context: Context): Long | Number | BigInt {
// TODO(pabloem): How do we deal with large integers?
return reader.uint64().low;
var shift = 0
var result = 0
const encoded: Uint8Array = BytesCoder.INSTANCE.decode(reader, context);
var i = 0;
while (true) {
const byte = encoded[i];
i++;
if (byte < 0) {
throw new Error('VarLong not terminated.')
}
const bits = byte & 0x7F
if (shift >= 64 || (shift >= 63 && bits > 1)) {
throw new Error('VarLong too long.')
}
const shuftedBits = bits << shift;
result = result | shuftedBits
shift += 7
if (!(byte & 0x80)) {
break
}
if (shift >= 32) {
// TODO(pabloem) support numbers larger than 1 << 32
// We are unable to decode further into javascript
break;
}
// TODO(pabloem): Remove this because it's giving us trouble!
// if (result >= 1 << 63) {
// result -= 1 << 64
// }
}
return result
}
}
CODER_REGISTRY.register(VarIntCoder.URN, VarIntCoder);
Expand All @@ -168,6 +246,7 @@ CODER_REGISTRY.register(DoubleCoder.URN, DoubleCoder);

export class BoolCoder extends FakeCoder<Boolean> {
static URN: string = "beam:coder:bool:v1";
type: string = "boolcoder";
encode(element: Boolean, writer: Writer, context: Context) {
writer.bool(element as boolean);
}
Expand All @@ -176,4 +255,4 @@ export class BoolCoder extends FakeCoder<Boolean> {
return reader.bool();
}
}
CODER_REGISTRY.register(BoolCoder.URN, BoolCoder);
CODER_REGISTRY.register(BoolCoder.URN, BoolCoder);
13 changes: 10 additions & 3 deletions sdks/node-ts/test/standard_coders_test.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,13 @@ import util = require('util');

const STANDARD_CODERS_FILE = '../../model/fn-execution/src/main/resources/org/apache/beam/model/fnexecution/v1/standard_coders.yaml';

const UNSUPPORTED_EXAMPLES = {
"beam:coder:varint:v1-7": "",
}

// TODO(pabloem): Empty this list.
const UNSUPPORTED_CODERS = [
"beam:coder:interval_window:v1",
"beam:coder:string_utf8:v1",
"beam:coder:double:v1",
"beam:coder:iterable:v1",
"beam:coder:timer:v1",
Expand Down Expand Up @@ -125,7 +128,8 @@ describe("standard Beam coders on Javascript", function() {
var context = (doc.nested === true) ? Context.needsDelimiters : Context.wholeStream;
const spec = doc;

const coder = CODER_REGISTRY.get(urn);
const coderConstructor = CODER_REGISTRY.get(urn);
const coder = new coderConstructor();
describeCoder(coder, urn, context, spec);
});
});
Expand All @@ -135,8 +139,11 @@ function describeCoder<T>(coder: Coder<T>, urn, context, spec: CoderSpec) {
let examples = 0;
const parser = get_json_value_parser(spec.coder);
for (let expected in spec.examples) {
var value = parser(spec.examples[expected]);
examples += 1;
if ((urn + '-' + examples) in UNSUPPORTED_EXAMPLES) {
continue;
}
var value = parser(spec.examples[expected]);
const expectedEncoded = Buffer.from(expected, 'binary')
coderCase(coder, value, expectedEncoded, context, examples);
}
Expand Down

0 comments on commit 884ad82

Please sign in to comment.