/* Online Auction system on ATLaS
Chang Richard Luo
lc@cs.ucla.edu

Carlo Zaniolo
zaniolo@cs.ucla.edu

http://wis.cs.ucla.edu/atlas
2003/5

reference:
http://www-db.stanford.edu/stream/sqr/onauc.html
*/

/* Category: Relation containing information about various categories (e.g., books, toys) of auctioned items. */
table Category(id int         /* unique identifier */,
         name char(100)       /* name of the category */,
         description char(100)/* description of the category */) memory; 



/*Item: Relation containing information about each item that is auctioned. */
table Item(id int              /* unique identifier */,
     name char(100)            /* name of the item */,
     description char(100)     /* description of the item */, 
     categoryID ref(Category)      /* category that this item belongs to. 
                         References Category.id  */,
     registrationTime int/* time when this item was registered */) memory; 



/* Person: Relation containing information about registered users who are allowed to start and bid for auctions. */
table Person(id  int             /* unique identifier */,
       name   char(100)          /* name of this person */,
       emailAddress    char(100) /* registered email address */,
       city   char(100)          /* city of residence */, 
       state  char(20)          /* state of residence */,
       registrationTime int/* time when this user registered */) memory; 



/* OpenAuction: Stream of openings of auctions. */
table OpenAuction(itemID  ref(Item)    /* id of the item being auctioned. Key attribute of
                           this stream. References Item.id */,
            sellerID  ref(Person)  /* seller of the item being auctioned. References
                           Person.id */,
            start_price int /* starting price of the item */, 
            timestamp   int /* time when the auction started */) memory; 



/* ClosedAuction: Stream of auction closings. */
table ClosedAuction(itemID     ref(Item)/* id of the item in this auction. Key attribute 
                            of this stream. References Item.id */,
              buyerID    ref(Person)/* buyer of this item. References Person.id. 
                            Could be NULL if there was no buyer. */,
              timestamp  int /* time when the auction closed */) memory; 



/* Bid: Stream of bids for items. A bid for itemID appears only after the OpenAuction tuple for itemID and before the ClosedAuction tuple for itemID. */
table Bid(itemID   ref(Item)  /* the item being bid for. References Item.id */,
    bid_price int /* bid price */,    
    bidderID  ref(Person) /* id the bidder. References Person.id */,
    timestamp int /* time when bid was registered */) memory; 

/* CurrentOpen: Table of currently open auction */
table CurrentOpen(itemID ref(Item), sellerID ref(Person)) memory;

/* AllPrice: Stream of union of OpenAuction, Bid, and ClosedAuction */
table AllPrice (itemID ref(Item),
		price int,
		timestamp int,
		tag char(10)) memory;

/* ClosedPrice: Stream of close price */
table ClosedPrice (itemID ref(Item),
		   price int,
		   timestamp int) memory;

/************* UDA *******************/
aggregate DolToEuro(price int):int
{
initialize:iterate:
{
	insert into return values(price*2);
}
};

aggregate deleteClose(d_id int):int
{
initialize:iterate:
{
	delete from CurrentOpen
		where itemID = d_id;
}
};

aggregate mymax(price int, tag char(20), iTimestamp int):(oPrice int, oTimestamp int)
{
table memo(mprice int, timestamp int) memory;
initialize:{
	insert into memo values(price, iTimestamp);
}
iterate:{
	update memo
	set mprice = price,
	    timestamp = iTimestamp
	where price < mprice;
	insert into return
		select mprice,timestamp from memo
	where tag = 'end';
}
};
AGGREGATE maxpair(iValue Int, iPoint int):(mValue Int, mPoint int)
{ TABLE mpair(value Int, point Int);
INITIALIZE: 
ITERATE: { 
	delete from mpair
	where iValue > value;
	insert into mpair values(iValue, iPoint)
	where exists(
		select value
		from mpair
		where iValue = value
	) OR 0=(select count(value) from mpair);
} 
REVISE:{
	/* delete expired tuples */
	delete from mpair 
	where iValue > value OR exists(
		select ex.iValue, ex.iPoint
		from expired ex
		where ex.iPoint = mpair.point
	);
	/* insert current tuple if necessary */
	insert into mpair values(iValue, iPoint)
	where exists(
		select value
		from mpair
		where iValue = value
	) OR 0=(select count(value) from mpair);
	/* output every 5 tuples */
	
}
TERMINATE: { 
	INSERT INTO RETURN
	SELECT value, point FROM mpair;
}
};

aggregate hotItem(itemID ref(Item)):(counter int, itemID ref(Item))
{
table counter(ct int) memory as values(0);
table memo(mItemID ref(Item), c int) memory;
initialize:iterate:{
	/* update the counter for current tuple */
	update memo
	set c = c + 1
	where itemID = mItemID;
	/* initialize the counter for brand-new tuple */
	insert into memo values(itemID, 1)
	where SQLCODE <>0;
}
revise:{
	/* increase the counter for the purpose of output */
	update counter
	set ct = (ct+1) % 5;
	/* delete the expired tuples */
	update memo
	set c = c - (select count(itemID) 
			from expired ex, memo
			where ex.itemID = mItemID);
	/* update the counter for current tuple */
	update memo
	set c = c + 1
	where itemID = mItemID;
	/* initialize the counter for brand-new tuple */
	insert into memo values(itemID, 1)
	where SQLCODE <>0;
	/* output every 5 tuples */
	insert into return 
		select maxpair(c, mItemID) 
		from counter, memo
		where ct % 5 = 0;
}
};

/********************************/

insert into Category values(0,'Book', 'Book desc'), (1, 'Notebook', 'Notebook desc');
/*
insert into Item values(0,'Algorithm', 'Computer Algorithm', 0, 0);
*/
insert into Item 
	select 0,'Algorithm', 'Computer Algorithm', c.OID, 0
	from Category c
	where c.name = 'Book';
insert into Item
	select 1, 'IBM ThinkPad', 'IBM TP desc', c.OID,0
	from Category c
	where name = 'Notebook';
insert into Person values(0,'Chang Luo', 'lc@cs.ucla.edu', 'Los Angeles', 'CA', 0);

insert into OpenAuction
	select i.OID, p.OID, 1, 0
	from Item i, Person p
	where i.id = 0 and p.id = 0;

insert into ClosedAuction
	select i.OID, p.OID, 10
	from Item i, Person p
	where i.id = 0 and p.id = 0;
insert into Bid
	select i.OID, 2, p.OID, 1
	from Item i, Person p
	where i.id = 0 and p.id = 0;
/*insert into Bid values(null,3,null,3), (null,4,null,4);*/

/* Ex 1. 
Currency Conversion Query: Convert the prices of incoming bids from U.S. dollars into Euros. 
Comments: This query relies on two CQL defaults: the default Unbounded window on Bid stream, and the default outermost Istream operator. DolToEuro is an user-defined function to convert a dollar price to euro. */
Select  itemID->name, DolToEuro(bid_price), bidderID->name
From    Bid
group by itemID->name, bidderID->name;

/* Ex 2.
Selection Query: Select all bids on a specified set of 5 items. 
*/

Select  itemID->name, bid_price
From    Bid
Where   itemID->id = 0 or itemID->id = 1020 or itemID->id = 2001 or itemID->id = 2019 or itemID->id = 1087;

/* Ex 3.
Local Item Suggestion Query: Report all items in category 10 that are put up for auction by some seller in Oregon. 
Comments: This query returns items that were put up for auction by some seller who lived in Oregon when the auction started. Changes in the seller's address after the start of an auction will not affect the result of this query. 

CQL:
Window -- last tuple
Select  P.name, P.city, O.itemID
From    OpenAuction [Now] O, Person P, Item I
Where   O.sellerID = P.id and P.state = 'OR' and O.itemID = I.id 
        and I.categoryID = 10 
*/

insert into stdout values('Local Item Suggestion Query: Report all items in category 10 that are put up for auction by some seller in Oregon. ');
select P.name, P.city, O.itemID->id
from OpenAuction O, Person P, Item I
where O.sellerID = P.id and P.state = 'OR' and O.itemID = I.OID and I.categoryID->id = 10;


/* Ex 4.
Open Auctions Query: Maintain a table of the currently open auctions. 

CQL:
Select * 
From   OpenAuction 
Where  itemID Not In (Select itemID 
                      From   ClosedAuction)

This CQL statement doesn't seems correct.  The result should be a table instead of a stream.

We should support:
insert into table
	select * from stream

Then it's user's responsibility to ensure the table won't grow infinitely.

Since ATLaS doesn't support stream in subquries, we use UDA instead in this example.
*/
insert into stdout values('Open Auctions Query: Maintain a table of the currently open auctions.');
insert into CurrentOpen
	select itemID, sellerID
	from OpenAuction;
select deleteClose(itemID)
from ClosedAuction;


/* Ex 5.
Closing Price Query: Report the closing price of each auction. 


CurrentPrice: 
    Select    P.itemID, Max(P.price) as price
    From      ((Select itemID, bid_price as price 
                From   Bid) Union All
               (Select itemID, start_price as price 
                From OpenAuction)) P
    Group By  P.itemID

Select   Rstream(C.itemID, P.price)
From     ClosedAuction [Now] C, CurrentPrice P
Where    C.itemID = P.itemID


Comments: We assume that the closing price in an auction is the price of the maximum bid, or the starting price of the auction in case there were no bids. 
*/
insert into stdout values('Closing Price Query: Report the closing price of each auction.');
insert into AllPrice
	select itemID, start_price, timestamp, 'start'
	from OpenAuction
	UNION ALL
	select itemID, bid_price, timestamp, 'bid'
	from Bid
	UNION ALL
	select itemID, -1, timestamp, 'end'	/* ClosedAuction doesn't have a price */
	from ClosedAuction;
insert into ClosedPrice
select itemID, mymax(price, tag, timestamp)
from AllPrice
group by itemID;


/* Ex 6
Average Closing Price Query: Monitor the average closing price across items in each category over the last hour. 
CurrentPrice: 
    Select    P.itemID, Max(P.price) as price
    From      ((Select itemID, bid_price as price 
                From   Bid) Union All
               (Select itemID, start_price as price 
                From OpenAuction)) P
    Group By  P.itemID
ClosingPriceStream:
    Select   Rstream(T.id as catID, P.price as price)
    From     ClosedAuction [Now] C, CurrentPrice P, 
             Item I, Category T    
    Where    C.itemID = P.itemID and C.itemID = I.id and I.categoryID = T.id 
AvgPrice:
    Select   catID, Avg(price)
    From     ClosingPriceStream [Range 1 Hour]
    Group By catID 
*/



/* Ex 7.
Short Auctions Query: Report all auctions that closed within five hours of their opening. 
Select Rstream(OpenAuction.*)
From   OpenAuction [Range 5 Hour] O, ClosedAuction [Now] C
Where  O.itemID = C.itemID
*/

/* Ex. 8
Hot Item Query: Select the item(s) with the most bids in the past hour. Update the results every minute. 


HotItemStream: 
    Select Rstream(itemID)
    From   (Select   B1.itemID as itemID, Count(*) as num
            From     Bid [Range 60 Minute
                          Slide 1 Minute] B1
            Group By B1.itemID) 
    Where   num >= All (Select   Count(*)
                        From     Bid [Range 60 Minute
                                      Slide 1 Minute] B2
                        Group By B2.itemID)

Select *
From   HotItemStream [Range 1 Minute]

*/

/* TODO: need time-based window
We change the question for tuple-based window:
Hot Item Query: Select the item(s) with the most bids in the past 10 bids.
Update the result every 5 bids.
*/


select itemID, hotItem(itemID)
#over (rows 10 preceding)
from Bid
group by itemID;


/* Ex 9.
Average Selling Price By Seller Query: For each seller, maintain the average selling price over the last 10 items sold. 
CurrentPrice: 
    Select    P.itemID, Max(P.price) as price
    From      ((Select itemID, bid_price as price 
                From   Bid) Union All
               (Select itemID, start_price as price 
                From OpenAuction)) P
    Group By  P.itemID
ClosingPriceStream:
    Select   Rstream(O.sellerID as sellerID, P.price as price)
    From     ClosedAuction [Now] C, CurrentPrice P, 
             OpenAuction O
    Where    C.itemID = P.itemID and C.itemID = O.itemID
AvgSellingPrice:
    Select   sellerID, Avg(price)
    From     ClosingPriceStream [Partition By sellerID Rows 10] 
    Group By sellerID 
*/
/*
select sellerID, avg(C.price)
over (partition by sellerID rows 10 preceding)
from ClosedPrice C, OpenAuction O
where O.itemID = C.itemID;
*/


/* Ex 10.
Highest Bid Query: Every 10 minutes return the highest bid(s) in the recent 10 minutes. 


Select  Rstream(itemID, bid_price)
From    Bid [Range 10 Minute
             Slide 10 Minute]
Where   bid_price = (Select  Max(bid_price)
                     From    Bid [Range 10 Minute]
                                  Slide 10 Minute]

TODO: need time-based window
We change the question to :
Highest Bid Query: Every 10 tuples return the highest bid(s) in the recent 10 tupels. 
*/
select maxpair(bid_price, itemID)
#over(rows 10 preceding)
from Bid;