class FileBackedElementCoder(Coder):
  def __init__(self, underlying_coder, root, threshold=1 << 20):
    self._underlying_coder = underlying_coder
    self._root = root
    self._threshold = threshold
  def encode(self, value):
    encoded = self._underlying_coder.encode(value)
    if len(encoded) < self._threshold:
      return b'\x00' + encoded
    else:
      path = beam.io.filesystems.FileSystems.join(self._root, uuid.uuid4().hex)
      with beam.io.filesystems.FileSystems.create(path) as fout:
        fout.write(encoded)
      return b'\x01' + path.encode('utf-8')
  def decode(self, encoded_value):
    if encoded_value[0] == 0:
      data = encoded_value[1:]
    else:
      path = encoded_value[1:].decode('utf-8')
      with beam.io.filesystems.FileSystems.open(path) as fin:
        data = fin.read(-1)
    return self._underlying_coder.decode(data)

import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.util.UUID;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.fs.ResolveOptions;
import org.apache.beam.sdk.io.fs.ResourceId;
public class HugeElementCoder<T> extends CustomCoder<T> {
  private final Coder<T> underlying;
  private final long threshold;
  private final ResourceId tempDir;
  public HugeElementCoder(Coder<T> underlying, long threshold, String tempDir) {
    this(underlying, threshold, FileSystems.matchNewResource(tempDir, true));
  }
  public HugeElementCoder(Coder<T> underlying, long threshold, ResourceId tempDir) {
    this.underlying = underlying;
    this.threshold = threshold;
    assert tempDir.isDirectory();
    this.tempDir = tempDir;
  }
  @Override
  public void encode(T value, OutputStream outStream) throws CoderException, IOException {
    MultiplexingOutputStream out = new MultiplexingOutputStream();
    try {
      underlying.encode(value, out);
    } finally {
      out.close(outStream);
    }
  }
  @Override
  public T decode(InputStream inStream) throws IOException {
    int isInline = inStream.read();
    if (isInline == 1) {
      return underlying.decode(inStream);
    } else {
      String path = new DataInputStream(inStream).readUTF();
      ReadableByteChannel fileChannel =
          FileSystems.open(
              tempDir.resolve(path, ResolveOptions.StandardResolveOptions.RESOLVE_FILE));
      try {
        return underlying.decode(Channels.newInputStream(fileChannel));
      } finally {
        fileChannel.close();
      }
    }
  }
  @Override
  public void verifyDeterministic() throws NonDeterministicException {
    this.underlying.verifyDeterministic();
  }
  private class MultiplexingOutputStream extends OutputStream {
    private boolean inlined;
    private ByteArrayOutputStream buffer;
    private OutputStream fileOut;
    private String path;
    public MultiplexingOutputStream() {
      this.inlined = true;
      this.buffer = new ByteArrayOutputStream();
    }
    @Override
    public void write(int b) throws IOException {
      write(new byte[] {(byte) b}, 0, 1);
    }
    @Override
    public void write(byte[] b) throws IOException {
      write(b, 0, b.length);
    }
    @Override
    public void write(byte[] b, int off, int len) throws IOException {
      if (inlined) {
        if (buffer.size() + len < threshold) {
          buffer.write(b, off, len);
        } else {
          inlined = false;
          path = UUID.randomUUID().toString();
          WritableByteChannel fileChannel =
              FileSystems.create(
                  tempDir.resolve(path, ResolveOptions.StandardResolveOptions.RESOLVE_FILE),
                  "application/octet-stream");
          fileOut = Channels.newOutputStream(fileChannel);
          fileOut.write(buffer.toByteArray());
          buffer = null;
          fileOut.write(b, off, len);
        }
      } else {
        fileOut.write(b, off, len);
      }
    }
    public void close(OutputStream out) throws IOException {
      if (inlined) {
        out.write(1);
        out.write(buffer.toByteArray());
      } else {
        fileOut.close();
        out.write(0);
        new DataOutputStream(out).writeUTF(path);
      }
    }
  }
}