/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.
 */
#ifndef READER_FILTER_OPERATOR_AND_FILTER_H
#define READER_FILTER_OPERATOR_AND_FILTER_H

#include "binary_filter.h"
// #include "storage/storage_utils.h"

namespace storage {

class AndFilter : public BinaryFilter {
   public:
    AndFilter() : BinaryFilter() {}
    ~AndFilter() {}
    AndFilter(Filter* left, Filter* right) : BinaryFilter(left, right) {}

    FORCE_INLINE bool satisfy(Statistic* statistic) {
        return left_->satisfy(statistic) && right_->satisfy(statistic);
    }

    FORCE_INLINE bool satisfy(int64_t time, int64_t value) {
        return left_->satisfy(time, value) && right_->satisfy(time, value);
    }

    FORCE_INLINE bool satisfy_start_end_time(int64_t start_time,
                                             int64_t end_time) {
        return left_->satisfy_start_end_time(start_time, end_time) &&
               right_->satisfy_start_end_time(start_time, end_time);
    }

    FORCE_INLINE bool contain_start_end_time(int64_t start_time,
                                             int64_t end_time) {
        return left_->contain_start_end_time(start_time, end_time) &&
               right_->contain_start_end_time(start_time, end_time);
    }

    std::vector<TimeRange*>* get_time_ranges() {
        std::vector<TimeRange*>* result = new std::vector<TimeRange*>();
        std::vector<TimeRange*>* left_time_ranges = left_->get_time_ranges();
        std::vector<TimeRange*>* right_time_ranges = right_->get_time_ranges();

        int left_index = 0, right_index = 0;
        int left_size = left_time_ranges->size();
        int right_size = right_time_ranges->size();

        while (left_index < left_size && right_index < right_size) {
            TimeRange* left_range = left_time_ranges->at(left_index);
            TimeRange* right_range = right_time_ranges->at(right_index);

            if (left_range->end_time_ < right_range->start_time_) {
                left_index++;
            } else if (right_range->end_time_ < left_range->start_time_) {
                right_index++;
            } else {
                TimeRange* intersection = new TimeRange(
                    std::max(left_range->start_time_, right_range->start_time_),
                    std::min(left_range->end_time_, right_range->end_time_));
                result->push_back(intersection);
                if (left_range->end_time_ <= intersection->end_time_) {
                    left_index++;
                }
                if (right_range->end_time_ <= intersection->end_time_) {
                    right_index++;
                }
            }
        }
        return result;
    }
};

}  // namespace storage

#endif  // READER_FILTER_OPERATOR_AND_FILTER_H
