<commit_before>/* Copyright 2016 <NAME>, NVIDIA Corporation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "scanner/api/database.h"
#include "scanner/engine/metadata.h"

#include "scanner/util/common.h"
#include "scanner/util/h264.h"
#include "scanner/util/util.h"

#include "storehouse/storage_backend.h"

#include <glog/logging.h>
#include <thread>

// For video
extern "C" {
#include "libavcodec/avcodec.h"
#include "libavfilter/avfilter.h"
#include "libavformat/avformat.h"
#include "libavformat/avio.h"
#include "libavutil/error.h"
#include "libavutil/opt.h"
#include "libavutil/pixdesc.h"
#include "libswscale/swscale.h"
}

#include <cassert>
#include <fstream>

using storehouse::StoreResult;
using storehouse::WriteFile;
using storehouse::RandomReadFile;

namespace scanner {
namespace internal {
namespace {

const std::string BAD_VIDEOS_FILE_PATH = "bad_videos.txt";

struct FFStorehouseState {
  std::unique_ptr<RandomReadFile> file = nullptr;
  size_t size = 0;  // total file size
  u64 pos = 0;
};

// For custom AVIOContext that loads from memory
i32 read_packet(void* opaque, u8* buf, i32 buf_size) {
  FFStorehouseState* fs = (FFStorehouseState*)opaque;
  size_t size_read;
  storehouse::StoreResult result;
  EXP_BACKOFF(fs->file->read(fs->pos, buf_size, buf, size_read), result);
  if (result != storehouse::StoreResult::EndOfFile) {
    exit_on_error(result);
  }

  fs->pos += size_read;
  return static_cast<i32>(size_read);
}

i64 seek(void* opaque, i64 offset, i32 whence) {
  FFStorehouseState* fs = (FFStorehouseState*)opaque;
  switch (whence) {
    case SEEK_SET:
      assert(offset >= 0);
      fs->pos = static_cast<u64>(offset);
      break;
    case SEEK_CUR:
      fs->pos += offset;
      break;
    case SEEK_END:
      fs->pos = fs->size;
      break;
    case AVSEEK_SIZE:
      return fs->size;
      break;
  }
  return fs->size - fs->pos;
}

struct CodecState {
  AVPacket av_packet;
  AVFrame* picture;
  AVFormatContext* format_context;
  AVIOContext* io_context;
  AVCodec* in_codec;
  AVCodecContext* in_cc;
#if LIBAVCODEC_VERSION_INT >= AV_VERSION_INT(57, 34, 0)
  AVCodecParameters* in_cc_params;
#endif
  i32 video_stream_index;
  AVBitStreamFilterContext* annexb;
};

bool setup_video_codec(FFStorehouseState* fs, CodecState& state) {
  VLOG(1) << "Setting up video codec";
  av_init_packet(&state.av_packet);
  state.picture = av_frame_alloc();
  state.format_context = avformat_alloc_context();

  size_t avio_context_buffer_size = 4096;
  u8* avio_context_buffer =
    static_cast<u8*>(av_malloc(avio_context_buffer_size));
  state.io_context =
    avio_alloc_context(avio_context_buffer, avio_context_buffer_size, 0, fs,
                       &read_packet, NULL, &seek);
  state.format_context->pb = state.io_context;

  // Read file header
  VLOG(1) << "Opening input file to read format";
  if (avformat_open_input(&state.format_context, NULL, NULL, NULL) < 0) {
    LOG(ERROR) << "open input failed";
    return false;
  }
  // Some formats don't have a header
  if (avformat_find_stream_info(state.format_context, NULL) < 0) {
    LOG(ERROR) << "find stream info failed";
    return false;
  }

  av_dump_format(state.format_context, 0, NULL, 0);

  // Find the best video stream in our input video
  state.video_stream_index = av_find_best_stream(
    state.format_context, AVMEDIA_TYPE_VIDEO, -1 /* auto select */,
    -1 /* no related stream */, &state.in_codec, 0 /* flags */);
  if (state.video_stream_index < 0) {
    LOG(ERROR) << "could not find best stream";
    return false;
  }

  AVStream const* const in_stream =
    state.format_context->streams[state.video_stream_index];

  state.in_codec = avcodec_find_decoder(AV_CODEC_ID_H264);
  if (state.in_codec == NULL) {
    LOG(FATAL) << "could not find h264 decoder";
  }

  state.in_cc = avcodec_alloc_context3(state.in_codec);
#if LIBAVCODEC_VERSION_INT >= AV_VERSION_INT(57, 34, 0)
  state.in_cc_params = avcodec_parameters_alloc();
  if (avcodec_parameters_from_context(state.in_cc_params, in_stream->codec) <
      0) {
    LOG(ERROR) << "could not copy codec params from input stream";
    return false;
  }
  if (avcodec_parameters_to_context(state.in_cc, state.in_cc_params) < 0) {
    LOG(ERROR) << "could not copy codec params to in cc";
    return false;
  }
#else
  if (avcodec_copy_context(state.in_cc, in_stream->codec) < 0) {
    LOG(ERROR) << "could not copy codec params to in cc";
    return false;
  }
#endif

  if (avcodec_open2(state.in_cc, state.in_codec, NULL) < 0) {
    LOG(ERROR) << "could not open codec";
    return false;
  }

  state.annexb = av_bitstream_filter_init("h264_mp4toannexb");

  return true;
}

void cleanup_video_codec(CodecState state) {
#if LIBAVCODEC_VERSION_INT >= AV_VERSION_INT(57, 34, 0)
  avcodec_parameters_free(&state.in_cc_params);
#endif
#if LIBAVCODEC_VERSION_INT >= AV_VERSION_INT(55, 53, 0)
  avcodec_free_context(&state.in_cc);
#else
  avcodec_close(state.in_cc);
  av_freep(&state.in_cc);
#endif
  avformat_close_input(&state.format_context);
  if (state.io_context) {
    av_freep(&state.io_context->buffer);
    av_freep(&state.io_context);
  }
  av_frame_free(&state.picture);
  av_bitstream_filter_close(state.annexb);
}

bool parse_and_write_video(storehouse::StorageBackend* storage,
                           const std::string& table_name, i32 table_id,
                           const std::string& path,
                           std::string& error_message) {
  proto::TableDescriptor table_desc;
  table_desc.set_id(table_id);
  table_desc.set_name(table_name);
  table_desc.set_job_id(-1);
  table_desc.set_timestamp(
    std::chrono::duration_cast<std::chrono::seconds>(now().time_since_epoch())
      .count());

  {
    Column* index_col = table_desc.add_columns();
    index_col->set_name(index_column_name());
    index_col->set_id(0);
    index_col->set_type(ColumnType::Other);

    Column* frame_col = table_desc.add_columns();
    frame_col->set_name(frame_column_name());
    frame_col->set_id(1);
    frame_col->set_type(ColumnType::Video);

    Column* frame_info_col = table_desc.add_columns();
    frame_info_col->set_name(frame_info_column_name());
    frame_info_col->set_id(2);
    frame_info_col->set_type(ColumnType::Other);
  }

  // Setup custom buffer for libavcodec so that we can read from a storehouse
  // file instead of a posix file
  FFStorehouseState file_state{};
  StoreResult result;
  EXP_BACKOFF(make_unique_random_read_file(storage, path, file_state.file),
              result);
  if (result != StoreResult::Success) {
    error_message = "Can not open video file";
    return false;
  }

  EXP_BACKOFF(file_state.file->get_size(file_state.size), result);
  if (result != StoreResult::Success) {
    error_message = "Can not get file size";
    return false;
  }
  if (file_state.size <= 0) {
    error_message = "Can not ingest empty video file";
    return false;
  }

  file_state.pos = 0;

  CodecState state;
  if (!setup_video_codec(&file_state, state)) {
    error_message = "Failed to set up video codec";
    return false;
  }

  VideoMetadata video_meta;
  proto::VideoDescriptor& video_descriptor = video_meta.get_descriptor();
  video_descriptor.set_table_id(table_id);
  video_descriptor.set_column_id(1);
  video_descriptor.set_item_id(0);

  video_descriptor.set_width(state.in_cc->width);
  video_descriptor.set_height(state.in_cc->height);
  video_descriptor.set_chroma_format(proto::VideoDescriptor::YUV_420);
  video_descriptor.set_codec_type(proto::VideoDescriptor::H264);

  std::string data_path = table_item_output_path(table_id, 1, 0);
  std::unique_ptr<WriteFile> demuxed_bytestream{};
  BACKOFF_FAIL(make_unique_write_file(storage, data_path, demuxed_bytestream));

  u64 bytestream_pos = 0;
  std::vector<u8> metadata_bytes;
  std::vector<i64> keyframe_positions;
  std::vector<i64> keyframe_timestamps;
  std::vector<i64> keyframe_byte_offsets;

  bool succeeded = true;
  i64 frame = 0;
  bool extradata_extracted = false;
  bool in_meta_packet_sequence = false;
  i64 meta_packet_sequence_start_offset = 0;
  bool saw_sps_nal = false;
  bool saw_pps_nal = false;
  std::map<u32, SPS> sps_map;
  std::map<u32, PPS> pps_map;
  u32 last_sps = -1;
  u32 last_pps = -1;
  std::map<u32, std::vector<u8>> sps_nal_bytes;
  std::map<u32, std::vector<u8>> pps_nal_bytes;
  SliceHeader prev_sh;

  i64 num_non_ref_frames = 0;
  i64 avcodec_frame = 0;
  while (true) {
    // Read from format context
    i32 err = av_read_frame(state.format_context, &state.av_packet);
    if (err == AVERROR_EOF) {
      av_packet_unref(&state.av_packet);
      break;
    } else if (err != 0) {
      char err_msg[256];
      av_strerror(err, err_msg, 256);
      LOG(ERROR) << "Error while decoding frame " << frame << " (" << err
                 << "): " << err_msg;
      cleanup_video_codec(state);
      error_message = "Error while decoding frame " + std::to_string(frame) +
                      " (" + std::to_string(err) + "): " + std::string(err_msg);
      return false;
    }

    if (state.av_packet.stream_index != state.video_stream_index) {
      av_packet_unref(&state.av_packet);
      continue;
    }

    /* NOTE1: some codecs are stream based (mpegvideo, mpegaudio)
       and this is the only method to use them because you cannot
       know the compressed data size before analysing it.

       BUT some other codecs (msmpeg4, mpeg4) are inherently frame
       based, so you must call them with all the data for one
       frame exactly. You must also initialize 'width' and
       'height' before initializing them. */

    /* NOTE2: some codecs allow the raw parameters (frame size,
       sample rate) to be changed at any frame. We handle this, so
       you should also take care of it */

    /* here, we use a stream based decoder (mpeg1video), so we
       feed decoder and see if it could decode a frame */

    u8* orig_data = state.av_packet.data;
    i32 orig_size = state.av_packet.size;

    u8* filtered_data;
    i32 filtered_data_size;
    if (av_bitstream_filter_filter(
          state.annexb, state.in_cc, NULL, &filtered_data, &filtered_data_size,
          state.av_packet.data, state.av_packet.size,
          state.av_packet.flags & AV_PKT_FLAG_KEY) < 0) {
      char err_msg[256];
      av_strerror(err, err_msg, 256);
      LOG(ERROR) << "Error while filtering " << frame << " (" << frame
                 << "): " << err_msg;
      cleanup_video_codec(state);
      error_message = "Error while filtering frame " + std::to_string(frame) +
                      " (" + std::to_string(err) + "): " + std::string(err_msg);
      return false;
    }

    if (!extradata_extracted) {
      const u8* extradata = state.in_cc->extradata;
      i32 extradata_size_left = state.in_cc->extradata_size;

      metadata_bytes.resize(extradata_size_left);
      memcpy(metadata_bytes.data(), extradata, extradata_size_left);

      while (extradata_size_left > 3) {
        const u8* nal_start = nullptr;
        i32 nal_size = 0;
        next_nal(extradata, extradata_size_left, nal_start, nal_size);
        i32 nal_ref_idc = (*nal_start >> 5);
        i32 nal_unit_type = (*nal_start) & 0x1F;
        VLOG(2) << "extradata nal size: " << nal_size << ", nal ref "
                << nal_ref_idc << ", nal unit " << nal_unit_type;
      }
      extradata_extracted = true;
    }

    i64 nal_bytestream_offset = bytestream_pos;

    VLOG(2) << "new packet " << nal_bytestream_offset;
    bool insert_sps_nal = false;
    // Parse NAL unit
    const u8* nal_parse = filtered_data;
    i32 size_left = filtered_data_size;
    i32 nals_parsed = 0;
    while (size_left > 3) {
      const u8* nal_start = nullptr;
      i32 nal_size = 0;
      next_nal(nal_parse, size_left, nal_start, nal_size);

      i32 nal_ref_idc = (*nal_start >> 5);
      i32 nal_unit_type = (*nal_start) & 0x1F;
      VLOG(2) << "frame " << frame << ", nal size " << nal_size
              << ", nal_ref_idc " << nal_ref_idc << ", nal unit "
              << nal_unit_type;
      if (nal_ref_idc == 0) {
        num_non_ref_frames += 1;
      }
      if (nal_unit_type > 4) {
        if (!in_meta_packet_sequence) {
          meta_packet_sequence_start_offset = nal_bytestream_offset;
          filtered_data_size - size_left;
          VLOG(2) << "in meta sequence " << nal_bytestream_offset;
          in_meta_packet_sequence = true;
          saw_sps_nal = false;
        }
      }
      std::vector<u8> rbsp_buffer;
      rbsp_buffer.reserve(64 * 1024);
      u32 consecutive_zeros = 0;
      i32 bytes = nal_size - 1;
      const u8* pb = nal_start + 1;
      while (bytes > 0) {
        /* Copy the byte into the rbsp, unless it
         * is the 0x03 in a 0x000003 */
        if (consecutive_zeros < 2 || *pb != 0x03) {
          rbsp_buffer.push_back(*pb);
        }
        if (*pb == 0) {
          ++consecutive_zeros;
        } else {
          consecutive_zeros = 0;
        }
        ++pb;
        --bytes;
      }

      // We need to track the last SPS NAL because some streams do
      // not insert an SPS every keyframe and we need to insert it
      // ourselves.
      // fprintf(stderr, "nal_size %d, rbsp size %lu\n", nal_size,
      // rbsp_buffer.size());
      const u8* rbsp_start = rbsp_buffer.data();
      i32 rbsp_size = rbsp_buffer.size();

      // SPS
      if (nal_unit_type == 7) {
        saw_sps_nal = true;
        i32 offset = 8;
        GetBitsState gb;
        gb.buffer = rbsp_start;
        gb.offset = 0;
        SPS sps;
        if (!parse_sps(gb, sps)) {
          error_message = "Failed to parse sps";
          return false;
        }
        i32 sps_id = sps.sps_id;
        sps_map[sps_id] = sps;
        last_sps = sps.sps_id;

        sps_nal_bytes[sps_id].clear();
        sps_nal_bytes[sps_id].insert(sps_nal_bytes[sps_id].end(), nal_start - 3,
                                     nal_start + nal_size + 3);
        VLOG(2) << "Last SPS NAL (" << sps_id << ", " << offset << ")"
                << " seen at frame " << frame;
      }
      // PPS
      if (nal_unit_type == 8) {
        GetBitsState gb;
        gb.buffer = rbsp_start;
        gb.offset = 0;
        PPS pps;
        if (!parse_pps(gb, pps)) {
          error_message = "Failed to parse pps";
          return false;
        }
        pps_map[pps.pps_id] = pps;
        last_pps = pps.pps_id;
        saw_pps_nal = true;
        i32 pps_id = pps.pps_id;
        pps_nal_bytes[pps_id].clear();
        pps_nal_bytes[pps_id].insert(pps_nal_bytes[pps_id].end(), nal_start - 3,
                                     nal_start + nal_size + 3);
        VLOG(2) << "PPS id " << pps.pps_id << ", SPS id " << pps.sps_id
                << ", frame " << frame;
      }
      if (is_vcl_nal(nal_unit_type)) {
        assert(last_pps != -1);
        assert(last_sps != -1);
        GetBitsState gb;
        gb.buffer = nal_start;
        gb.offset = 8;
        SliceHeader sh;
        if (!parse_slice_header(gb, sps_map.at(last_sps), pps_map,
                                nal_unit_type, nal_ref_idc, sh)) {
          error_message = "Failed to parse slice header";
          return false;
        }
        if (frame == 0 || is_new_access_unit(sps_map, pps_map, prev_sh, sh)) {
          frame++;
          size_t bytestream_offset;
          if (nal_unit_type == 5) {
            // Insert an SPS NAL if we did not see one in the meta packet
            // sequence
            keyframe_byte_offsets.push_back(nal_bytestream_offset);
            keyframe_positions.push_back(frame - 1);
            keyframe_timestamps.push_back(state.av_packet.pts);
            saw_sps_nal = false;
            VLOG(2) << "keyframe " << frame - 1 << ", byte offset "
                    << meta_packet_sequence_start_offset;

            // Insert metadata
            VLOG(2) << "inserting sps and pss nals";<commit_msg>Add read cache for ingest<commit_after>/* Copyright 2016 <NAME>, NVIDIA Corporation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "scanner/api/database.h"
#include "scanner/engine/metadata.h"

#include "scanner/util/common.h"
#include "scanner/util/h264.h"
#include "scanner/util/util.h"

#include "storehouse/storage_backend.h"

#include <glog/logging.h>
#include <thread>

// For video
extern "C" {
#include "libavcodec/avcodec.h"
#include "libavfilter/avfilter.h"
#include "libavformat/avformat.h"
#include "libavformat/avio.h"
#include "libavutil/error.h"
#include "libavutil/opt.h"
#include "libavutil/pixdesc.h"
#include "libswscale/swscale.h"
}

#include <cassert>
#include <fstream>

using storehouse::StoreResult;
using storehouse::WriteFile;
using storehouse::RandomReadFile;

namespace scanner {
namespace internal {
namespace {

const std::string BAD_VIDEOS_FILE_PATH = "bad_videos.txt";

struct FFStorehouseState {
  std::unique_ptr<RandomReadFile> file = nullptr;
  size_t size = 0;  // total file size
  u64 pos = 0;

  u64 buffer_start = 0;
  u64 buffer_end = 0;
  std::vector<u8> buffer;
};

// For custom AVIOContext that loads from memory
i32 read_packet(void* opaque, u8* buf, i32 buf_size) {
  FFStorehouseState* fs = (FFStorehouseState*)opaque;
  if (!(fs->buffer_start <= fs->pos &&
        fs->pos + buf_size < fs->buffer_end)) {
    // Not in cache
    size_t buffer_size = 64 * 1024 * 1024;
    fs->buffer.resize(buffer_size);
    size_t size_read;
    storehouse::StoreResult result;
    EXP_BACKOFF(
      fs->file->read(fs->pos, buffer_size, fs->buffer.data(), size_read),
      result);
    if (result != storehouse::StoreResult::EndOfFile) {
      exit_on_error(result);
    }

    fs->buffer_start = fs->pos;
    fs->buffer_end = fs->pos + size_read;
  }

  size_t size_read = std::min((size_t)buf_size, fs->buffer_end - fs->pos);
  memcpy(buf, fs->buffer.data() + (fs->pos - fs->buffer_start), size_read);
  fs->pos += size_read;
  return static_cast<i32>(size_read);
}

i64 seek(void* opaque, i64 offset, i32 whence) {
  FFStorehouseState* fs = (FFStorehouseState*)opaque;
  switch (whence) {
    case SEEK_SET:
      assert(offset >= 0);
      fs->pos = static_cast<u64>(offset);
      break;
    case SEEK_CUR:
      fs->pos += offset;
      break;
    case SEEK_END:
      fs->pos = fs->size;
      break;
    case AVSEEK_SIZE:
      return fs->size;
      break;
  }
  return fs->size - fs->pos;
}

struct CodecState {
  AVPacket av_packet;
  AVFrame* picture;
  AVFormatContext* format_context;
  AVIOContext* io_context;
  AVCodec* in_codec;
  AVCodecContext* in_cc;
#if LIBAVCODEC_VERSION_INT >= AV_VERSION_INT(57, 34, 0)
  AVCodecParameters* in_cc_params;
#endif
  i32 video_stream_index;
  AVBitStreamFilterContext* annexb;
};

bool setup_video_codec(FFStorehouseState* fs, CodecState& state) {
  VLOG(1) << "Setting up video codec";
  av_init_packet(&state.av_packet);
  state.picture = av_frame_alloc();
  state.format_context = avformat_alloc_context();

  size_t avio_context_buffer_size = 4096;
  u8* avio_context_buffer =
    static_cast<u8*>(av_malloc(avio_context_buffer_size));
  state.io_context =
    avio_alloc_context(avio_context_buffer, avio_context_buffer_size, 0, fs,
                       &read_packet, NULL, &seek);
  state.format_context->pb = state.io_context;

  // Read file header
  VLOG(1) << "Opening input file to read format";
  if (avformat_open_input(&state.format_context, NULL, NULL, NULL) < 0) {
    LOG(ERROR) << "open input failed";
    return false;
  }
  // Some formats don't have a header
  if (avformat_find_stream_info(state.format_context, NULL) < 0) {
    LOG(ERROR) << "find stream info failed";
    return false;
  }

  av_dump_format(state.format_context, 0, NULL, 0);

  // Find the best video stream in our input video
  state.video_stream_index = av_find_best_stream(
    state.format_context, AVMEDIA_TYPE_VIDEO, -1 /* auto select */,
    -1 /* no related stream */, &state.in_codec, 0 /* flags */);
  if (state.video_stream_index < 0) {
    LOG(ERROR) << "could not find best stream";
    return false;
  }

  AVStream const* const in_stream =
    state.format_context->streams[state.video_stream_index];

  state.in_codec = avcodec_find_decoder(AV_CODEC_ID_H264);
  if (state.in_codec == NULL) {
    LOG(FATAL) << "could not find h264 decoder";
  }

  state.in_cc = avcodec_alloc_context3(state.in_codec);
#if LIBAVCODEC_VERSION_INT >= AV_VERSION_INT(57, 34, 0)
  state.in_cc_params = avcodec_parameters_alloc();
  if (avcodec_parameters_from_context(state.in_cc_params, in_stream->codec) <
      0) {
    LOG(ERROR) << "could not copy codec params from input stream";
    return false;
  }
  if (avcodec_parameters_to_context(state.in_cc, state.in_cc_params) < 0) {
    LOG(ERROR) << "could not copy codec params to in cc";
    return false;
  }
#else
  if (avcodec_copy_context(state.in_cc, in_stream->codec) < 0) {
    LOG(ERROR) << "could not copy codec params to in cc";
    return false;
  }
#endif

  if (avcodec_open2(state.in_cc, state.in_codec, NULL) < 0) {
    LOG(ERROR) << "could not open codec";
    return false;
  }

  state.annexb = av_bitstream_filter_init("h264_mp4toannexb");

  return true;
}

void cleanup_video_codec(CodecState state) {
#if LIBAVCODEC_VERSION_INT >= AV_VERSION_INT(57, 34, 0)
  avcodec_parameters_free(&state.in_cc_params);
#endif
#if LIBAVCODEC_VERSION_INT >= AV_VERSION_INT(55, 53, 0)
  avcodec_free_context(&state.in_cc);
#else
  avcodec_close(state.in_cc);
  av_freep(&state.in_cc);
#endif
  avformat_close_input(&state.format_context);
  if (state.io_context) {
    av_freep(&state.io_context->buffer);
    av_freep(&state.io_context);
  }
  av_frame_free(&state.picture);
  av_bitstream_filter_close(state.annexb);
}

bool parse_and_write_video(storehouse::StorageBackend* storage,
                           const std::string& table_name, i32 table_id,
                           const std::string& path,
                           std::string& error_message) {
  proto::TableDescriptor table_desc;
  table_desc.set_id(table_id);
  table_desc.set_name(table_name);
  table_desc.set_job_id(-1);
  table_desc.set_timestamp(
    std::chrono::duration_cast<std::chrono::seconds>(now().time_since_epoch())
      .count());

  {
    Column* index_col = table_desc.add_columns();
    index_col->set_name(index_column_name());
    index_col->set_id(0);
    index_col->set_type(ColumnType::Other);

    Column* frame_col = table_desc.add_columns();
    frame_col->set_name(frame_column_name());
    frame_col->set_id(1);
    frame_col->set_type(ColumnType::Video);

    Column* frame_info_col = table_desc.add_columns();
    frame_info_col->set_name(frame_info_column_name());
    frame_info_col->set_id(2);
    frame_info_col->set_type(ColumnType::Other);
  }

  // Setup custom buffer for libavcodec so that we can read from a storehouse
  // file instead of a posix file
  FFStorehouseState file_state{};
  StoreResult result;
  EXP_BACKOFF(make_unique_random_read_file(storage, path, file_state.file),
              result);
  if (result != StoreResult::Success) {
    error_message = "Can not open video file";
    return false;
  }

  EXP_BACKOFF(file_state.file->get_size(file_state.size), result);
  if (result != StoreResult::Success) {
    error_message = "Can not get file size";
    return false;
  }
  if (file_state.size <= 0) {
    error_message = "Can not ingest empty video file";
    return false;
  }

  file_state.pos = 0;

  CodecState state;
  if (!setup_video_codec(&file_state, state)) {
    error_message = "Failed to set up video codec";
    return false;
  }

  VideoMetadata video_meta;
  proto::VideoDescriptor& video_descriptor = video_meta.get_descriptor();
  video_descriptor.set_table_id(table_id);
  video_descriptor.set_column_id(1);
  video_descriptor.set_item_id(0);

  video_descriptor.set_width(state.in_cc->width);
  video_descriptor.set_height(state.in_cc->height);
  video_descriptor.set_chroma_format(proto::VideoDescriptor::YUV_420);
  video_descriptor.set_codec_type(proto::VideoDescriptor::H264);

  std::string data_path = table_item_output_path(table_id, 1, 0);
  std::unique_ptr<WriteFile> demuxed_bytestream{};
  BACKOFF_FAIL(make_unique_write_file(storage, data_path, demuxed_bytestream));

  u64 bytestream_pos = 0;
  std::vector<u8> metadata_bytes;
  std::vector<i64> keyframe_positions;
  std::vector<i64> keyframe_timestamps;
  std::vector<i64> keyframe_byte_offsets;

  bool succeeded = true;
  i64 frame = 0;
  bool extradata_extracted = false;
  bool in_meta_packet_sequence = false;
  i64 meta_packet_sequence_start_offset = 0;
  bool saw_sps_nal = false;
  bool saw_pps_nal = false;
  std::map<u32, SPS> sps_map;
  std::map<u32, PPS> pps_map;
  u32 last_sps = -1;
  u32 last_pps = -1;
  std::map<u32, std::vector<u8>> sps_nal_bytes;
  std::map<u32, std::vector<u8>> pps_nal_bytes;
  SliceHeader prev_sh;

  i64 num_non_ref_frames = 0;
  i64 avcodec_frame = 0;
  while (true) {
    // Read from format context
    i32 err = av_read_frame(state.format_context, &state.av_packet);
    if (err == AVERROR_EOF) {
      av_packet_unref(&state.av_packet);
      break;
    } else if (err != 0) {
      char err_msg[256];
      av_strerror(err, err_msg, 256);
      LOG(ERROR) << "Error while decoding frame " << frame << " (" << err
                 << "): " << err_msg;
      cleanup_video_codec(state);
      error_message = "Error while decoding frame " + std::to_string(frame) +
                      " (" + std::to_string(err) + "): " + std::string(err_msg);
      return false;
    }

    if (state.av_packet.stream_index != state.video_stream_index) {
      av_packet_unref(&state.av_packet);
      continue;
    }

    /* NOTE1: some codecs are stream based (mpegvideo, mpegaudio)
       and this is the only method to use them because you cannot
       know the compressed data size before analysing it.

       BUT some other codecs (msmpeg4, mpeg4) are inherently frame
       based, so you must call them with all the data for one
       frame exactly. You must also initialize 'width' and
       'height' before initializing them. */

    /* NOTE2: some codecs allow the raw parameters (frame size,
       sample rate) to be changed at any frame. We handle this, so
       you should also take care of it */

    /* here, we use a stream based decoder (mpeg1video), so we
       feed decoder and see if it could decode a frame */

    u8* orig_data = state.av_packet.data;
    i32 orig_size = state.av_packet.size;

    u8* filtered_data;
    i32 filtered_data_size;
    if (av_bitstream_filter_filter(
          state.annexb, state.in_cc, NULL, &filtered_data, &filtered_data_size,
          state.av_packet.data, state.av_packet.size,
          state.av_packet.flags & AV_PKT_FLAG_KEY) < 0) {
      char err_msg[256];
      av_strerror(err, err_msg, 256);
      LOG(ERROR) << "Error while filtering " << frame << " (" << frame
                 << "): " << err_msg;
      cleanup_video_codec(state);
      error_message = "Error while filtering frame " + std::to_string(frame) +
                      " (" + std::to_string(err) + "): " + std::string(err_msg);
      return false;
    }

    if (!extradata_extracted) {
      const u8* extradata = state.in_cc->extradata;
      i32 extradata_size_left = state.in_cc->extradata_size;

      metadata_bytes.resize(extradata_size_left);
      memcpy(metadata_bytes.data(), extradata, extradata_size_left);

      while (extradata_size_left > 3) {
        const u8* nal_start = nullptr;
        i32 nal_size = 0;
        next_nal(extradata, extradata_size_left, nal_start, nal_size);
        i32 nal_ref_idc = (*nal_start >> 5);
        i32 nal_unit_type = (*nal_start) & 0x1F;
        VLOG(2) << "extradata nal size: " << nal_size << ", nal ref "
                << nal_ref_idc << ", nal unit " << nal_unit_type;
      }
      extradata_extracted = true;
    }

    i64 nal_bytestream_offset = bytestream_pos;

    VLOG(2) << "new packet " << nal_bytestream_offset;
    bool insert_sps_nal = false;
    // Parse NAL unit
    const u8* nal_parse = filtered_data;
    i32 size_left = filtered_data_size;
    i32 nals_parsed = 0;
    while (size_left > 3) {
      const u8* nal_start = nullptr;
      i32 nal_size = 0;
      next_nal(nal_parse, size_left, nal_start, nal_size);

      i32 nal_ref_idc = (*nal_start >> 5);
      i32 nal_unit_type = (*nal_start) & 0x1F;
      VLOG(2) << "frame " << frame << ", nal size " << nal_size
              << ", nal_ref_idc " << nal_ref_idc << ", nal unit "
              << nal_unit_type;
      if (nal_ref_idc == 0) {
        num_non_ref_frames += 1;
      }
      if (nal_unit_type > 4) {
        if (!in_meta_packet_sequence) {
          meta_packet_sequence_start_offset = nal_bytestream_offset;
          filtered_data_size - size_left;
          VLOG(2) << "in meta sequence " << nal_bytestream_offset;
          in_meta_packet_sequence = true;
          saw_sps_nal = false;
        }
      }
      std::vector<u8> rbsp_buffer;
      rbsp_buffer.reserve(64 * 1024);
      u32 consecutive_zeros = 0;
      i32 bytes = nal_size - 1;
      const u8* pb = nal_start + 1;
      while (bytes > 0) {
        /* Copy the byte into the rbsp, unless it
         * is the 0x03 in a 0x000003 */
        if (consecutive_zeros < 2 || *pb != 0x03) {
          rbsp_buffer.push_back(*pb);
        }
        if (*pb == 0) {
          ++consecutive_zeros;
        } else {
          consecutive_zeros = 0;
        }
        ++pb;
        --bytes;
      }

      // We need to track the last SPS NAL because some streams do
      // not insert an SPS every keyframe and we need to insert it
      // ourselves.
      // fprintf(stderr, "nal_size %d, rbsp size %lu\n", nal_size,
      // rbsp_buffer.size());
      const u8* rbsp_start = rbsp_buffer.data();
      i32 rbsp_size = rbsp_buffer.size();

      // SPS
      if (nal_unit_type == 7) {
        saw_sps_nal = true;
        i32 offset = 8;
        GetBitsState gb;
        gb.buffer = rbsp_start;
        gb.offset = 0;
        SPS sps;
        if (!parse_sps(gb, sps)) {
          error_message = "Failed to parse sps";
          return false;
        }
        i32 sps_id = sps.sps_id;
        sps_map[sps_id] = sps;
        last_sps = sps.sps_id;

        sps_nal_bytes[sps_id].clear();
        sps_nal_bytes[sps_id].insert(sps_nal_bytes[sps_id].end(), nal_start - 3,
                                     nal_start + nal_size + 3);
        VLOG(2) << "Last SPS NAL (" << sps_id << ", " << offset << ")"
                << " seen at frame " << frame;
      }
      // PPS
      if (nal_unit_type == 8) {
        GetBitsState gb;
        gb.buffer = rbsp_start;
        gb.offset = 0;
        PPS pps;
        if (!parse_pps(gb, pps)) {
          error_message = "Failed to parse pps";
          return false;
        }
        pps_map[pps.pps_id] = pps;
        last_pps = pps.pps_id;
        saw_pps_nal = true;
        i32 pps_id = pps.pps_id;
        pps_nal_bytes[pps_id].clear();
        pps_nal_bytes[pps_id].insert(pps_nal_bytes[pps_id].end(), nal_start - 3,
                                     nal_start + nal_size + 3);
        VLOG(2) << "PPS id " << pps.pps_id << ", SPS id " << pps.sps_id
                << ", frame " << frame;
      }
      if (is_vcl_nal(nal_unit_type)) {
        assert(last_pps != -1);
        assert(last_sps != -1);
        GetBitsState gb;
        gb.buffer = nal_start;
        gb.offset = 8;
        SliceHeader sh;
        if (!parse_slice_header(gb, sps_map.at(last_sps), pps_map,
                                nal_unit_type, nal_ref_idc, sh)) {
          error_message = "Failed to parse slice header";
          return false;
        }
        //printf("ref_idx_l0 %d, ref_idx_l1 %d\n",
        // sh.num_ref_idx_l0_active, sh.num_ref_idx_l1_active);
        if (frame == 0 || is_new_access_unit(sps_map, pps_map, prev_sh, sh)) {
          frame++;
          size_t bytestream_offset;
          if (nal_unit_type == 5) {
            // Insert an SPS NAL if we did not see one in the meta packet
            // sequence
            keyframe_byte_offsets.push_back(nal_bytestream_offset);
            keyframe_positions.push_back(frame - 1);
            keyframe_timestamps.push_back(state.av_packet.pts);
            saw_sps_nal = false;
            VLOG(2) << "keyframe " << frame - 1 << ", byte offset "
                    << meta_packet_sequence_start_offset;

            // Insert metadata
            VLOG(2) << "inserting sps and pss nals";
{'fraction_non_alphanumeric': 0.1165161753273934, 'fraction_numerical': 0.01689290199126951, 'mean_word_length': 2.940968540120184, 'pattern_counts': {'":': 0, '<': 302, '<?xml version=': 0, '>': 149, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2455705', 'n_tokens_mistral': 12081, 'n_tokens_neox': 11544, 'n_words': 2578}