https://speakerdeck.com/qnighy/parsing-protobuf-in-bigquery

protobuf-example.sql

CREATE TEMP FUNCTION parseMessage(messageB64 BYTES)
RETURNS STRING
LANGUAGE js AS """
  const toSextet = (ch) => {
    if (0x41 <= ch && ch < 0x41 + 26) {
      return ch - 0x41;
    } else if (0x61 <= ch && ch < 0x71 + 26) {
      return ch - (0x61 - 26);
    } else if (0x30 <= ch && ch < 0x30 + 10) {
      return ch + (52 - 0x30);
    } else if (ch === 0x2B) {
      return 62;
    } else if (ch === 0x2F) {
      return 63;
    } else {
      return 0;
    }
  };
  const decodeBase64 = (message) => {
    const numPadded =
      message.length < 2 ? 0 :
      message.charCodeAt(message.length - 2) === 0x3D ? 2 :
      message.charCodeAt(message.length - 1) === 0x3D ? 1 : 0;
    const numChunks = (message.length / 4) | 0;
    const ret = new Array(numChunks * 3 - numPadded);
    for(let i = 0; i < numChunks; ++i) {
      const chunkPad = i + 1 == numChunks ? numPadded : 0;
      const ch0 = toSextet(message.charCodeAt(i * 4));
      const ch1 = toSextet(message.charCodeAt(i * 4 + 1));
      const ch2 = toSextet(message.charCodeAt(i * 4 + 2));
      const ch3 = toSextet(message.charCodeAt(i * 4 + 3));
      ret[i * 3] = (ch0 << 2) | (ch1 >> 4);
      if (chunkPad < 2) ret[i * 3 + 1] = ((ch1 & 15) << 4) | (ch2 >> 2);
      if (chunkPad < 1) ret[i * 3 + 2] = ((ch2 & 3) << 6) | ch3;
    }
    return ret;
  };
  class Parser {
    constructor(bytes, index, limit) {
      this.bytes = bytes;
      this.index = index;
      this.limit = limit;
    }
    
    eatByte() {
      if (this.index >= this.limit) {
        throw `EOF: ${this.index} + 1 > ${this.limit}`;
      }
      return this.bytes[this.index++];
    }
    
    eatBytes(len) {
      const index = this.index;
      if (this.index + len > this.limit) {
        throw `EOF: ${this.index} + ${len} > ${this.limit}`;
      }
      this.index += len;
      return this.bytes.slice(index, index + len);
    }
    
    subParser(len) {
      const index = this.index;
      if (index + len > this.limit) {
        throw `EOF: ${this.index} + ${len} > ${this.limit}`;
      }
      this.index += len;
      return new Parser(this.bytes, index, index + len);
    }
    
    convertToBytes() {
      const bytes = this.bytes.slice(this.index, this.limit);
      this.index = this.limit;
      return bytes;
    }
    
    convertToString() {
      const bytes = this.convertToBytes();
      const bytesString = String.fromCharCode(...bytes);
      return decodeURIComponent(escape(bytesString));
    }
    
    eatVarint() {
      let shift = 0;
      let ret = 0;
      while(true) {
        const byte = this.eatByte();
        ret |= (byte & 127) << shift;
        if ((byte & 128) == 0) {
          return ret;
        }
        shift += 7;
      }
    }
    
    eatField() {
      const tag = this.eatVarint();
      const field = tag >> 3;
      const wireType = tag & 7;
      let ret;
      if (wireType === 0) {
        ret = this.eatVarint();
      } else if (wireType === 1) {
        ret = this.eatBytes(64);
      } else if (wireType === 2) {
        ret = this.subParser(this.eatVarint());
      } else if (wireType === 3) {
        throw "TODO: wireType == 3";
      } else if (wireType === 4) {
        throw "TODO: wireType == 4";
      } else if (wireType === 5) {
        ret = this.eatBytes(32);
      } else {
        throw "Unknown wire type";
      }
      return [field, wireType, ret];
    }
    
    parseFieldMask() {
      const ret = { paths: [] };
      while (this.index < this.limit) {
        const [field, wireType, fieldData] = this.eatField();
        if (field === 1) {
          ret.paths.push(fieldData.convertToString());
        }
      }
      return ret;
    }

    parseWorkingHistory() {
      const ret = {};
      while (this.index < this.limit) {
        const [field, wireType, fieldData] = this.eatField();
        if (field === 1) {
          ret._op = ["OPERATION_UNSPECIFIED", "CREATE", "UPDATE", "DELETE"][fieldData] || fieldData;
        } else if (field === 2) {
          ret.id = fieldData;
        }
      }
      return ret;
    }

    parseProfile() {
      const ret = { working_histories: [] };
      while (this.index < this.limit) {
        const [field, wireType, fieldData] = this.eatField();
        if (field === 1) {
          ret._op = ["OPERATION_UNSPECIFIED", "CREATE", "UPDATE", "DELETE"][fieldData] || fieldData;
        } else if (field === 2) {
          ret.id = fieldData;
        } else if (field === 17) {
          ret.working_histories.push(fieldData.parseWorkingHistory());
        }
      }
      return ret;
    }

    parseUser() {
      const ret = {};
      while (this.index < this.limit) {
        const [field, wireType, fieldData] = this.eatField();
        if (field === 1) {
          ret._op = ["OPERATION_UNSPECIFIED", "CREATE", "UPDATE", "DELETE"][fieldData] || fieldData;
        } else if (field === 2) {
          ret.id = fieldData;
        } else if (field === 9) {
          ret.profile = fieldData.parseProfile();
        }
      }
      return ret;
    }
    
    parseMessage() {
      const ret = {};
      while (this.index < this.limit) {
        const [field, wireType, fieldData] = this.eatField();
        if (field === 1) {
          ret.field_mask = fieldData.parseFieldMask();
        } else if (field === 2) {
          ret.user = fieldData.parseUser();
        }
      }
      return ret;
    }
  }
  const messageBytes = decodeBase64(messageB64);
  const parser = new Parser(messageBytes, 0, messageBytes.length);
  // try {
  return JSON.stringify(parser.parseMessage());
  // } catch(e) { return JSON.stringify({ error: e }); }
""";

CREATE TEMP FUNCTION should_perform(message_json STRING)
RETURNS BOOL
LANGUAGE js AS """
  const fetchCreatedWorkingHistories = (user, field_mask) => {
    if(!field_mask.paths.find((x) => x === "profile.working_histories._op")) return [];
    return user.profile.working_histories.filter((w) => w._op === "CREATE");
  };
  const { user, field_mask } = JSON.parse(message_json);
  return user.id && fetchCreatedWorkingHistories(user, field_mask).length > 0;
""";